taolib.testing.config_center

taolib.config_center - 中心化配置管理系统。

提供多环境配置管理、版本控制、审计日志和实时推送功能。

主要组件:

• models: 数据模型（配置、版本、审计日志、用户/角色）

• repository: 数据访问层（MongoDB）

• cache: 缓存层（Redis）

• services: 业务逻辑层

• validation: 配置验证框架

• server: Web 服务器（FastAPI）

• events: 事件系统

• client: 客户端 SDK

Submodules

• taolib.testing.config_center.cache
• taolib.testing.config_center.client
• taolib.testing.config_center.events
• taolib.testing.config_center.models
• taolib.testing.config_center.repository
• taolib.testing.config_center.server
• taolib.testing.config_center.services
• taolib.testing.config_center.validation

Attributes

__version__

Classes

ConfigCenterClient	配置中心客户端。
AuditLogResponse	审计日志响应模型。
ConfigCreate	创建配置请求模型。
ConfigResponse	配置响应模型。
ConfigUpdate	更新配置请求模型。
AuditAction	审计操作类型枚举。
AuditStatus	审计操作状态枚举。
ChangeType	配置变更类型枚举。
ConfigStatus	配置状态枚举。
ConfigValueType	配置值类型枚举。
Environment	环境类型枚举。
Permission	权限模型。
RoleCreate	创建角色请求模型。
RoleResponse	角色响应模型。
UserCreate	创建用户请求模型。
UserResponse	用户响应模型。
ConfigVersionResponse	配置版本响应模型。
ConfigValidator	配置验证器协议。
ValidationResult	验证结果。
ValidatorRegistry	验证器注册表。

Package Contents

taolib.testing.config_center.__version__

class taolib.testing.config_center.ConfigCenterClient(base_url: str, token: str, cache_ttl: int = 60)

配置中心客户端。

提供同步和异步方式获取配置，以及 WebSocket 监听配置变更。

_base_url

_token

_cache_ttl = 60

_local_cache: dict[str, tuple[Any, float]]

_headers

_cache_key(environment: str, service: str, key: str) → str

生成本地缓存 Key。

_get_cached(key: str) → Any | None

获取本地缓存。

_set_cached(key: str, value: Any) → None

设置本地缓存。

get_config(key: str, environment: str, service: str) → Any | None

同步获取配置。

参数:

• key -- 配置键

• environment -- 环境类型

• service -- 服务名称

返回:

配置值，如果不存在则返回 None

async aget_config(key: str, environment: str, service: str) → Any | None

异步获取配置。

参数:

• key -- 配置键

• environment -- 环境类型

• service -- 服务名称

返回:

配置值，如果不存在则返回 None

get_configs(environment: str, service: str) → list[dict[str, Any]]

获取服务的所有配置。

参数:

• environment -- 环境类型

• service -- 服务名称

返回:

配置列表

async watch_config(key: str, environment: str, service: str, callback: collections.abc.Callable[[dict[str, Any]], None]) → None

监听配置变更（需要 websockets 依赖）。

参数:

• key -- 配置键

• environment -- 环境类型

• service -- 服务名称

• callback -- 变更回调函数

class taolib.testing.config_center.AuditLogResponse

Bases: AuditLogBase

审计日志响应模型。

id: str

status: taolib.testing.config_center.models.enums.AuditStatus

timestamp: datetime.datetime

model_config

class taolib.testing.config_center.ConfigCreate

Bases: ConfigBase

创建配置请求模型。

class taolib.testing.config_center.ConfigResponse

Bases: ConfigBase

配置响应模型。

id: str

version: int

created_by: str

updated_by: str

created_at: datetime.datetime

updated_at: datetime.datetime

model_config

class taolib.testing.config_center.ConfigUpdate

Bases: pydantic.BaseModel

更新配置请求模型。

value: Any | None

value_type: taolib.testing.config_center.models.enums.ConfigValueType | None

description: str | None

schema_id: str | None

tags: list[str] | None

status: taolib.testing.config_center.models.enums.ConfigStatus | None

class taolib.testing.config_center.AuditAction

Bases: enum.StrEnum

审计操作类型枚举。

CONFIG_CREATE = 'config.create'

CONFIG_UPDATE = 'config.update'

CONFIG_DELETE = 'config.delete'

CONFIG_PUBLISH = 'config.publish'

CONFIG_ROLLBACK = 'config.rollback'

USER_LOGIN = 'user.login'

USER_LOGOUT = 'user.logout'

ROLE_ASSIGN = 'role.assign'

class taolib.testing.config_center.AuditStatus

Bases: enum.StrEnum

审计操作状态枚举。

SUCCESS = 'success'

FAILED = 'failed'

class taolib.testing.config_center.ChangeType

Bases: enum.StrEnum

配置变更类型枚举。

CREATE = 'create'

UPDATE = 'update'

DELETE = 'delete'

ROLLBACK = 'rollback'

class taolib.testing.config_center.ConfigStatus

Bases: enum.StrEnum

配置状态枚举。

DRAFT = 'draft'

ACTIVE = 'active'

DEPRECATED = 'deprecated'

class taolib.testing.config_center.ConfigValueType

Bases: enum.StrEnum

配置值类型枚举。

STRING = 'string'

NUMBER = 'number'

BOOLEAN = 'boolean'

JSON = 'json'

SECRET = 'secret'

class taolib.testing.config_center.Environment

Bases: enum.StrEnum

环境类型枚举。

DEVELOPMENT = 'development'

STAGING = 'staging'

PRE_PRODUCTION = 'pre-production'

PRODUCTION = 'production'

class taolib.testing.config_center.Permission

Bases: pydantic.BaseModel

权限模型。

resource: str

actions: list[str]

class taolib.testing.config_center.RoleCreate

Bases: RoleBase

创建角色请求模型。

class taolib.testing.config_center.RoleResponse

Bases: RoleBase

角色响应模型。

id: str

is_system: bool

created_at: datetime.datetime

updated_at: datetime.datetime

model_config

class taolib.testing.config_center.UserCreate

Bases: UserBase

创建用户请求模型。

password: str

role_ids: list[str]

class taolib.testing.config_center.UserResponse

Bases: UserBase

用户响应模型。

id: str

role_ids: list[str]

last_login: datetime.datetime | None

created_at: datetime.datetime

updated_at: datetime.datetime

model_config

class taolib.testing.config_center.ConfigVersionResponse

Bases: ConfigVersionBase

配置版本响应模型。

id: str

is_rollback_target: bool

created_at: datetime.datetime

model_config

class taolib.testing.config_center.ConfigValidator

Bases: Protocol

配置验证器协议。

所有验证器必须实现此协议。

validate(key: str, value: Any, context: dict[str, Any] | None = None) → ValidationResult

验证配置值。

参数:

• key -- 配置键

• value -- 配置值

• context -- 额外上下文信息

返回:

验证结果

class taolib.testing.config_center.ValidationResult

验证结果。

valid

验证是否通过

errors

错误消息列表

valid: bool

errors: list[str] = []

class taolib.testing.config_center.ValidatorRegistry

验证器注册表。

按配置键模式注册验证器，支持通配符匹配。

_validators: list[tuple[str, taolib.testing.config_center.validation.base.ConfigValidator]] = []

register(pattern: str, validator: taolib.testing.config_center.validation.base.ConfigValidator) → None

注册验证器。

参数:

• pattern -- 配置键模式（支持通配符，如 database.*）

• validator -- 验证器实例

get_validators(key: str) → list[taolib.testing.config_center.validation.base.ConfigValidator]

获取匹配配置键的所有验证器。

参数:

key -- 配置键

返回:

验证器列表

validate(key: str, value: Any, context: dict[str, Any] | None = None) → taolib.testing.config_center.validation.base.ValidationResult

执行验证链。

参数:

• key -- 配置键

• value -- 配置值

• context -- 额外上下文

返回:

聚合的验证结果