taolib.testing.data_sync.pipeline.protocols

Pipeline Protocol 定义。

定义 Extractor, Transformer, Loader 的接口契约。

Classes

TransformContext	转换上下文。
TransformResult	转换结果。
LoadResult	加载结果。
ExtractorProtocol	数据提取器 Protocol。
TransformerProtocol	数据转换器 Protocol。
LoaderProtocol	数据加载器 Protocol。
ValidateResult	验证结果。
ValidatorProtocol	数据验证器 Protocol。

Module Contents

class taolib.testing.data_sync.pipeline.protocols.TransformContext

转换上下文。

job_id: str

collection_name: str

field_mapping: dict[str, str] | None = None

transform_fn: Any | None = None

class taolib.testing.data_sync.pipeline.protocols.TransformResult

转换结果。

transformed: list[dict[str, Any]]

failures: list[dict[str, Any]] = []

class taolib.testing.data_sync.pipeline.protocols.LoadResult

加载结果。

inserted: int

updated: int

failed: int

failures: list[dict[str, Any]] = []

class taolib.testing.data_sync.pipeline.protocols.ExtractorProtocol

Bases: Protocol

数据提取器 Protocol。

async extract(source_collection: motor.motor_asyncio.AsyncIOMotorCollection, checkpoint: taolib.testing.data_sync.models.checkpoint.SyncCheckpoint | None, filter_query: dict[str, Any] | None, batch_size: int) → Any

分批提取源数据。

参数:

• source_collection -- 源集合

• checkpoint -- 检查点（增量同步时使用）

• filter_query -- 过滤查询

• batch_size -- 批次大小

生成器:

文档批次列表

class taolib.testing.data_sync.pipeline.protocols.TransformerProtocol

Bases: Protocol

数据转换器 Protocol。

async transform(documents: list[dict[str, Any]], context: TransformContext) → TransformResult

转换文档批次。

参数:

• documents -- 文档列表

• context -- 转换上下文

返回:

转换结果（成功文档 + 失败记录）

class taolib.testing.data_sync.pipeline.protocols.LoaderProtocol

Bases: Protocol

数据加载器 Protocol。

async load(target_collection: motor.motor_asyncio.AsyncIOMotorCollection, documents: list[dict[str, Any]]) → LoadResult

加载文档到目标集合。

参数:

• target_collection -- 目标集合

• documents -- 文档列表

返回:

加载结果

class taolib.testing.data_sync.pipeline.protocols.ValidateResult

验证结果。

valid: list[dict[str, Any]]

failures: list[dict[str, Any]] = []

class taolib.testing.data_sync.pipeline.protocols.ValidatorProtocol

Bases: Protocol

数据验证器 Protocol。

async validate(documents: list[dict[str, Any]], context: TransformContext) → ValidateResult

验证文档批次。

参数:

• documents -- 文档列表

• context -- 转换上下文

返回:

验证结果（通过的文档 + 失败记录）