taolib.testing.config_center.repository.base

泛型 Repository 基类模块。

向后兼容重导出：实际实现已迁移至 taolib._base.repository。

Classes

AsyncRepository

异步 Repository 基类。

Module Contents

class taolib.testing.config_center.repository.base.AsyncRepository(collection: motor.motor_asyncio.AsyncIOMotorCollection, model_class: type[T])

Bases: abc.ABC, Generic[T]

异步 Repository 基类。

_collection

_model_class

async create(document: dict[str, Any]) → T

创建文档。

参数:

document -- 文档数据字典

返回:

创建的文档对应的 Pydantic 模型实例

async get_by_id(doc_id: str) → T | None

根据 ID 获取文档。

参数:

doc_id -- 文档 ID

返回:

Pydantic 模型实例，如果不存在则返回 None

async update(doc_id: str, updates: dict[str, Any]) → T | None

更新文档。

参数:

• doc_id -- 文档 ID

• updates -- 更新字段字典

返回:

更新后的 Pydantic 模型实例，如果不存在则返回 None

async delete(doc_id: str) → bool

删除文档。

参数:

doc_id -- 文档 ID

返回:

是否删除成功

async list(filters: dict[str, Any] | None = None, skip: int = 0, limit: int = 100, sort: list[tuple[str, int]] | None = None) → list[T]

查询文档列表。

参数:

• filters -- 过滤条件

• skip -- 跳过记录数

• limit -- 限制记录数

• sort -- 排序条件

返回:

Pydantic 模型实例列表

async count(filters: dict[str, Any] | None = None) → int

统计文档数量。

参数:

filters -- 过滤条件

返回:

符合条件的文档数量