Skip to content

CacheConfig

Typed configuration powered by Pydantic for backend selection and options.

Bases: BaseModel

缓存配置类

使用 Pydantic 进行自动验证,采用 backend + options 的扩展模式。

支持从多种来源加载: - 字典 - YAML 文件 - TOML 文件 - JSON 文件 - 环境变量

配置示例 
# 方式1: 使用默认内存后端
config = CacheConfig()

# 方式2: 指定后端名称
config = CacheConfig(backend="memory")

# 方式3: 指定后端和参数
config = CacheConfig(backend="redis", options={"host": "localhost", "port": 6379})

# 实例化后端
backend = config.create_backend()
属性
Source code in src/symphra_cache/config.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
class CacheConfig(BaseModel):
    """
    缓存配置类

    使用 Pydantic 进行自动验证,采用 backend + options 的扩展模式。

    支持从多种来源加载:
    - 字典
    - YAML 文件
    - TOML 文件
    - JSON 文件
    - 环境变量

    配置示例:
        ```python
        # 方式1: 使用默认内存后端
        config = CacheConfig()

        # 方式2: 指定后端名称
        config = CacheConfig(backend="memory")

        # 方式3: 指定后端和参数
        config = CacheConfig(backend="redis", options={"host": "localhost", "port": 6379})

        # 实例化后端
        backend = config.create_backend()
        ```

    属性:
        backend: 已注册的后端名称(如 memory、file、redis)
        options: 传递给后端构造函数的参数字典
    """

    backend: str = Field(
        default="memory",
        description="后端名称,对应已注册的 backend 标识(如 memory、file、redis)",
    )

    options: dict[str, Any] = Field(
        default_factory=dict,
        description="后端构造参数,会在实例化时传递给具体后端实现",
    )

    # ========== Pydantic 配置 ==========

    model_config = {
        "validate_assignment": True,
        "extra": "forbid",  # 禁止额外字段
        "str_strip_whitespace": True,
    }

    # ========== 验证器 ==========

    @model_validator(mode="after")
    def validate_backend(self) -> CacheConfig:
        """验证后端配置"""
        from .backends import get_registered_backends

        # 验证后端类型
        available_backends = get_registered_backends()
        if self.backend.lower() not in available_backends:
            valid_backends = ", ".join(available_backends)
            msg = f"不支持的后端类型: {self.backend}。支持的类型: {valid_backends}"
            raise ValueError(msg)

        return self

    # ========== 后端实例化 ==========

    def create_backend(self) -> BaseBackend:
        """
        根据配置创建后端实例

        Returns:
            配置好的后端实例

        Raises:
            CacheConfigError: 后端创建失败

        示例:
            >>> config = CacheConfig(backend="memory", options={"max_size": 1000})
            >>> backend = config.create_backend()
        """
        from .backends import create_backend

        try:
            return create_backend(self.backend, **self.options)
        except Exception as e:
            msg = f"创建 {self.backend} 后端失败: {e}"
            raise CacheConfigError(msg) from e

    # ========== 工厂方法 ==========

    @classmethod
    def from_file(cls, file_path: str | Path) -> CacheConfig:
        """
        从配置文件创建配置

        支持的格式:
        - YAML (.yaml, .yml)
        - TOML (.toml)
        - JSON (.json)

        Args:
            file_path: 配置文件路径

        Returns:
            CacheConfig 实例

        Raises:
            CacheConfigError: 文件读取或解析失败

        示例:
            >>> config = CacheConfig.from_file("config/cache.yaml")
        """
        file_path = Path(file_path)

        if not file_path.exists():
            msg = f"配置文件不存在: {file_path}"
            raise CacheConfigError(msg)

        suffix = file_path.suffix.lower()

        try:
            if suffix in {".yaml", ".yml"}:
                return cls._from_yaml(file_path)
            if suffix == ".toml":
                return cls._from_toml(file_path)
            if suffix == ".json":
                return cls._from_json(file_path)
            msg = f"不支持的配置文件格式: {suffix}"
            raise CacheConfigError(msg)
        except Exception as e:
            if isinstance(e, CacheConfigError):
                raise
            msg = f"读取配置文件失败: {file_path}"
            raise CacheConfigError(msg) from e

    @classmethod
    def _from_yaml(cls, file_path: Path) -> CacheConfig:
        """从 YAML 文件加载"""
        try:
            import yaml
        except ImportError as e:
            msg = "YAML 支持需要安装 PyYAML: pip install pyyaml"
            raise ImportError(msg) from e

        with file_path.open("r", encoding="utf-8") as f:
            data = yaml.safe_load(f)

        if not isinstance(data, dict):
            msg = "YAML 配置文件必须是字典格式"
            raise CacheConfigError(msg)

        return cls(**data)

    @classmethod
    def _from_toml(cls, file_path: Path) -> CacheConfig:
        """从 TOML 文件加载"""
        try:
            import tomllib
        except ImportError:
            try:
                import tomli as tomllib  # Python < 3.11
            except ImportError as e:
                msg = "TOML 支持需要安装 tomli: pip install tomli"
                raise ImportError(msg) from e

        with file_path.open("rb") as f:
            data = tomllib.load(f)

        if not isinstance(data, dict):
            msg = "TOML 配置文件必须是字典格式"
            raise CacheConfigError(msg)

        return cls(**data)

    @classmethod
    def _from_json(cls, file_path: Path) -> CacheConfig:
        """从 JSON 文件加载"""
        import json

        with file_path.open("r", encoding="utf-8") as f:
            data = json.load(f)

        if not isinstance(data, dict):
            msg = "JSON 配置文件必须是字典格式"
            raise CacheConfigError(msg)

        return cls(**data)

    @classmethod
    def from_env(cls, prefix: str = "SYMPHRA_CACHE_") -> CacheConfig:
        """
        从环境变量创建配置

        环境变量命名规则:
        - SYMPHRA_CACHE_BACKEND=memory
        - SYMPHRA_CACHE_OPTIONS__MAX_SIZE=10000  (options 使用双下划线)
        - SYMPHRA_CACHE_OPTIONS__HOST=localhost

        Args:
            prefix: 环境变量前缀

        Returns:
            CacheConfig 实例

        示例:
            >>> import os
            >>> os.environ["SYMPHRA_CACHE_BACKEND"] = "redis"
            >>> os.environ["SYMPHRA_CACHE_OPTIONS__HOST"] = "localhost"
            >>> config = CacheConfig.from_env()
        """
        backend = os.environ.get(f"{prefix}BACKEND", "memory")
        options: dict[str, Any] = {}

        # 收集 options
        options_prefix = f"{prefix}OPTIONS__"
        for key, value in os.environ.items():
            if key.startswith(options_prefix):
                # 移除前缀并转为小写
                option_key = key[len(options_prefix) :].lower()
                # 类型转换
                options[option_key] = cls._convert_env_value(value)

        return cls(backend=backend, options=options)

    @staticmethod
    def _convert_env_value(value: str) -> Any:
        """转换环境变量值类型"""
        # 布尔值
        if value.lower() in {"true", "1", "yes", "on"}:
            return True
        if value.lower() in {"false", "0", "no", "off"}:
            return False

        # None/null
        if value.lower() in {"none", "null", ""}:
            return None

        # 数字
        try:
            if "." in value:
                return float(value)
            return int(value)
        except ValueError:
            pass

        # 字符串
        return value

    def __repr__(self) -> str:
        """字符串表示"""
        return f"CacheConfig(backend={self.backend!r}, options={self.options!r})"

__repr__()

字符串表示

Source code in src/symphra_cache/config.py 
285
286
287
def __repr__(self) -> str:
    """字符串表示"""
    return f"CacheConfig(backend={self.backend!r}, options={self.options!r})"

create_backend()

根据配置创建后端实例

Returns:

Type Description
BaseBackend

配置好的后端实例

Raises:

Type Description
CacheConfigError

后端创建失败
示例

config = CacheConfig(backend="memory", options={"max_size": 1000}) backend = config.create_backend()

Source code in src/symphra_cache/config.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def create_backend(self) -> BaseBackend:
    """
    根据配置创建后端实例

    Returns:
        配置好的后端实例

    Raises:
        CacheConfigError: 后端创建失败

    示例:
        >>> config = CacheConfig(backend="memory", options={"max_size": 1000})
        >>> backend = config.create_backend()
    """
    from .backends import create_backend

    try:
        return create_backend(self.backend, **self.options)
    except Exception as e:
        msg = f"创建 {self.backend} 后端失败: {e}"
        raise CacheConfigError(msg) from e

from_env(prefix='SYMPHRA_CACHE_') classmethod

从环境变量创建配置

环境变量命名规则: - SYMPHRA_CACHE_BACKEND=memory - SYMPHRA_CACHE_OPTIONS__MAX_SIZE=10000 (options 使用双下划线) - SYMPHRA_CACHE_OPTIONS__HOST=localhost

Parameters:

Name Type Description Default
prefix str

环境变量前缀

 'SYMPHRA_CACHE_'

Returns:

Type Description
CacheConfig

CacheConfig 实例
示例

import os os.environ["SYMPHRA_CACHE_BACKEND"] = "redis" os.environ["SYMPHRA_CACHE_OPTIONS__HOST"] = "localhost" config = CacheConfig.from_env()

Source code in src/symphra_cache/config.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
@classmethod
def from_env(cls, prefix: str = "SYMPHRA_CACHE_") -> CacheConfig:
    """
    从环境变量创建配置

    环境变量命名规则:
    - SYMPHRA_CACHE_BACKEND=memory
    - SYMPHRA_CACHE_OPTIONS__MAX_SIZE=10000  (options 使用双下划线)
    - SYMPHRA_CACHE_OPTIONS__HOST=localhost

    Args:
        prefix: 环境变量前缀

    Returns:
        CacheConfig 实例

    示例:
        >>> import os
        >>> os.environ["SYMPHRA_CACHE_BACKEND"] = "redis"
        >>> os.environ["SYMPHRA_CACHE_OPTIONS__HOST"] = "localhost"
        >>> config = CacheConfig.from_env()
    """
    backend = os.environ.get(f"{prefix}BACKEND", "memory")
    options: dict[str, Any] = {}

    # 收集 options
    options_prefix = f"{prefix}OPTIONS__"
    for key, value in os.environ.items():
        if key.startswith(options_prefix):
            # 移除前缀并转为小写
            option_key = key[len(options_prefix) :].lower()
            # 类型转换
            options[option_key] = cls._convert_env_value(value)

    return cls(backend=backend, options=options)

from_file(file_path) classmethod

从配置文件创建配置

支持的格式: - YAML (.yaml, .yml) - TOML (.toml) - JSON (.json)

Parameters:

Name Type Description Default
file_path str | Path

配置文件路径

 required

Returns:

Type Description
CacheConfig

CacheConfig 实例

Raises:

Type Description
CacheConfigError

文件读取或解析失败
示例

config = CacheConfig.from_file("config/cache.yaml")

Source code in src/symphra_cache/config.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@classmethod
def from_file(cls, file_path: str | Path) -> CacheConfig:
    """
    从配置文件创建配置

    支持的格式:
    - YAML (.yaml, .yml)
    - TOML (.toml)
    - JSON (.json)

    Args:
        file_path: 配置文件路径

    Returns:
        CacheConfig 实例

    Raises:
        CacheConfigError: 文件读取或解析失败

    示例:
        >>> config = CacheConfig.from_file("config/cache.yaml")
    """
    file_path = Path(file_path)

    if not file_path.exists():
        msg = f"配置文件不存在: {file_path}"
        raise CacheConfigError(msg)

    suffix = file_path.suffix.lower()

    try:
        if suffix in {".yaml", ".yml"}:
            return cls._from_yaml(file_path)
        if suffix == ".toml":
            return cls._from_toml(file_path)
        if suffix == ".json":
            return cls._from_json(file_path)
        msg = f"不支持的配置文件格式: {suffix}"
        raise CacheConfigError(msg)
    except Exception as e:
        if isinstance(e, CacheConfigError):
            raise
        msg = f"读取配置文件失败: {file_path}"
        raise CacheConfigError(msg) from e

validate_backend()

验证后端配置

Source code in src/symphra_cache/config.py 
87
88
89
90
91
92
93
94
95
96
97
98
99
@model_validator(mode="after")
def validate_backend(self) -> CacheConfig:
    """验证后端配置"""
    from .backends import get_registered_backends

    # 验证后端类型
    available_backends = get_registered_backends()
    if self.backend.lower() not in available_backends:
        valid_backends = ", ".join(available_backends)
        msg = f"不支持的后端类型: {self.backend}。支持的类型: {valid_backends}"
        raise ValueError(msg)

    return self