diff --git a/docs/library_integration_guide.md b/docs/library_integration_guide.md index 7c5e049..179f64f 100644 --- a/docs/library_integration_guide.md +++ b/docs/library_integration_guide.md @@ -11,7 +11,7 @@ 推荐接入方式： - `sync_state_machine` 负责同步引擎、状态机、pipeline、datasource 抽象。 -- 业务系统负责提供自己的数据源适配层，不要反向让 `sync_state_machine` 依赖业务系统。 +- 业务系统负责提供自己的 datasource / handler 适配层，不要反向让 `sync_state_machine` 依赖业务系统。 --- @@ -46,13 +46,18 @@ ecm-sync-run --config_path=run_profiles/preset/jsonl_to_api.default.yaml - `sync_state_machine.build_config` - `sync_state_machine.build_default_config` -- `sync_state_machine.run_pipeline_from_config` - `sync_state_machine.create_pipeline_from_config` +- `sync_state_machine.build_config_from_file` - `sync_state_machine.PipelineRunConfig` - `sync_state_machine.JsonlDataSourceConfig` - `sync_state_machine.ApiDataSourceConfig` - `sync_state_machine.DomainRegistry` +配置加载分两步： + +1. 先构造或读取 `PipelineRunConfig` +2. 再交给 `create_pipeline_from_config()` 创建 pipeline + 内置 domain 注册会在导入包时自动完成。 --- @@ -200,7 +205,7 @@ class DemoJsonlHandler(BaseJsonlHandler): --- -### 6.2 如果 backend 是数据库：建议写“数据库 handler / adapter”，不要让引擎直接依赖 ORM +### 6.2 如果 backend 是数据库：建议写 handler / adapter，不要让引擎直接依赖 ORM 推荐结构： @@ -234,6 +239,7 @@ backend/ - 不要在 `sync_state_machine` 包里 import backend 的 model/repository - 不要让 `DomainRegistry` 注册逻辑散落在 backend 业务层之外 +- 不要在 datasource 之外做额外的 project 过滤；`target_project_ids` 只应由 datasource 自己解释 --- @@ -268,6 +274,7 @@ backend/ - `strategy_class` - `jsonl_handler_class` - `api_handler_class` +- 以及任意通过 `register_handler()` 追加的 datasource handler 示例： @@ -282,9 +289,11 @@ DomainRegistry.register( jsonl_handler_class=DemoJsonlHandler, api_handler_class=DemoApiHandler, ) + +DomainRegistry.register_handler("demo", "custom_remote", DemoCustomHandler) ``` -如果是 backend 接入，建议在 backend 的 integration 启动阶段集中注册，而不是分散在 controller / service 里。 +如果是 backend 接入，建议在启动入口集中注册，而不是分散在 controller / service 里。 --- @@ -320,7 +329,7 @@ async def run_demo_sync() -> None: await pipeline.run() ``` -如果 backend 自己提供本地数据库数据，不要把 backend model 直接灌进 pipeline；先经过 handler / mapper 转成 schema 再进入引擎。 + 如果 backend 自己提供本地数据库数据，不要把 backend model 直接灌进 pipeline；先经过 handler / mapper 转成 schema 再进入引擎。 --- @@ -353,8 +362,9 @@ async def run_demo_sync() -> None: ## 11. 相关文件 - 包入口：[sync_state_machine/__init__.py](../sync_state_machine/__init__.py) -- CLI：[sync_state_machine/cli.py](../sync_state_machine/cli.py) +- CLI / 运行入口：[sync_state_machine/cli.py](../sync_state_machine/cli.py) - 配置工厂：[sync_state_machine/config](../sync_state_machine/config) - datasource 基础层：[sync_state_machine/datasource](../sync_state_machine/datasource) - pipeline 工厂：[sync_state_machine/pipeline/factory.py](../sync_state_machine/pipeline/factory.py) -- domain 注册：[sync_state_machine/common/registry.py](../sync_state_machine/common/registry.py) \ No newline at end of file +- domain 注册：[sync_state_machine/common/registry.py](../sync_state_machine/common/registry.py) +- backend 集成入口：[backend/app/thirdparty/ecm_sync/services/sync_runner.py](../backend/app/thirdparty/ecm_sync/services/sync_runner.py) \ No newline at end of file diff --git a/scripts/auto_test_domains/base.py b/scripts/auto_test_domains/base.py index 7835b45..02271bc 100644 --- a/scripts/auto_test_domains/base.py +++ b/scripts/auto_test_domains/base.py @@ -365,7 +365,7 @@ def run_pipeline_round( merge_tree_contents(common_dataset_dir, paths["temp_data_dir"]) enrich_supplier_display_fields( paths["temp_data_dir"], - f"{config['project_root']}/filtered_datasource/datasource/filtered/region_1.jsonl", + f"{config['project_root']}/working_local_datasource/filtered/region_1.jsonl", ) normalize_contract_settlement_records(paths["temp_data_dir"]) normalize_construction_task_plan_nodes(paths["temp_data_dir"]) diff --git a/sync_state_machine/common/registry.py b/sync_state_machine/common/registry.py index 7d5340e..d5e303c 100644 --- a/sync_state_machine/common/registry.py +++ b/sync_state_machine/common/registry.py @@ -17,7 +17,7 @@ Domain 注册中心 node_type="contract", schema=ContractResponse, node_class=ContractSyncNode, - jsonl_handler_class=ContractJsonlHandler + jsonl_handler_class=ContractJsonlHandler, ) # Collection 查询 SyncNode 类 @@ -47,7 +47,6 @@ class DomainRegistration: strategy_class: Optional[Type[Any]] = None # 可选，部分 domain 可能只读 jsonl_handler_class: Optional[Type[Any]] = None api_handler_class: Optional[Type[Any]] = None - db_handler_class: Optional[Type[Any]] = None handler_classes: Dict[str, Type[Any]] = field(default_factory=dict) metadata: Dict[str, Any] = field(default_factory=dict) # 额外元数据 @@ -69,7 +68,6 @@ class DomainRegistry: _BUILTIN_HANDLER_ATTRS: Dict[str, str] = { "jsonl": "jsonl_handler_class", "api": "api_handler_class", - "db": "db_handler_class", } @classmethod @@ -81,7 +79,6 @@ class DomainRegistry: strategy_class: Optional[Type[Any]] = None, jsonl_handler_class: Optional[Type[Any]] = None, api_handler_class: Optional[Type[Any]] = None, - db_handler_class: Optional[Type[Any]] = None, **metadata ) -> None: """ @@ -94,7 +91,6 @@ class DomainRegistry: strategy_class: Strategy 子类（可选） jsonl_handler_class: JSONL Handler 类（可选） api_handler_class: API Handler 类（可选） - db_handler_class: DB Handler 类（可选） **metadata: 额外的元数据 Raises: @@ -110,7 +106,6 @@ class DomainRegistry: strategy_class=strategy_class, jsonl_handler_class=jsonl_handler_class, api_handler_class=api_handler_class, - db_handler_class=db_handler_class, metadata=metadata ) for datasource_type, attr_name in cls._BUILTIN_HANDLER_ATTRS.items(): @@ -187,10 +182,6 @@ class DomainRegistry: setattr(reg, attr_name, handler_class) @classmethod - def register_db_handler(cls, node_type: str, db_handler_class: Type[Any]) -> None: - """为已注册 domain 补充 DB Handler。""" - cls.register_handler(node_type, "db", db_handler_class) - @classmethod def get_schema(cls, node_type: str) -> Type[BaseModel]: """获取 Schema 类""" @@ -245,7 +236,6 @@ class DomainRegistry: print(f" Strategy Class: {reg.strategy_class.__name__ if reg.strategy_class else '❌ None'}") print(f" JSONL Handler: {reg.jsonl_handler_class.__name__ if reg.jsonl_handler_class else '❌ None'}") print(f" API Handler: {reg.api_handler_class.__name__ if reg.api_handler_class else '❌ None'}") - print(f" DB Handler: {reg.db_handler_class.__name__ if reg.db_handler_class else '❌ None'}") extra_handler_types = sorted( handler_type for handler_type in reg.handler_classes if handler_type not in cls._BUILTIN_HANDLER_ATTRS ) @@ -272,7 +262,7 @@ class DomainRegistry: "has_strategy": True, "has_jsonl_handler": True, "has_api_handler": True, - "has_db_handler": False + "handler_types": [] }, ... } @@ -284,7 +274,6 @@ class DomainRegistry: "has_strategy": reg.strategy_class is not None, "has_jsonl_handler": reg.jsonl_handler_class is not None, "has_api_handler": reg.api_handler_class is not None, - "has_db_handler": reg.db_handler_class is not None, "handler_types": sorted(reg.handler_classes.keys()), } return summary diff --git a/tests/unit/test_datasource_type_registry.py b/tests/unit/test_datasource_type_registry.py index cab4bc1..5efedb5 100644 --- a/tests/unit/test_datasource_type_registry.py +++ b/tests/unit/test_datasource_type_registry.py @@ -76,7 +76,7 @@ class MemoryHandler(BaseNodeHandler[MemorySchema]): return [] -class MemoryRemoteDbHandler(BaseNodeHandler[MemorySchema]): +class MemoryRemoteHandler(BaseNodeHandler[MemorySchema]): __test__ = False def __init__(self, datasource: BaseDataSource): @@ -102,8 +102,7 @@ def _ensure_memory_domain_registered() -> None: reg.schema = MemorySchema reg.node_class = MemoryNode reg.strategy_class = MemoryStrategy - reg.db_handler_class = MemoryRemoteDbHandler - reg.handler_classes["db"] = MemoryRemoteDbHandler + reg.handler_classes["memory_remote"] = MemoryRemoteHandler reg.handler_classes["memory"] = MemoryHandler return @@ -112,9 +111,9 @@ def _ensure_memory_domain_registered() -> None: schema=MemorySchema, node_class=MemoryNode, strategy_class=MemoryStrategy, - db_handler_class=MemoryRemoteDbHandler, ) DomainRegistry.register_handler("memory_demo", "memory", MemoryHandler) + DomainRegistry.register_handler("memory_demo", "memory_remote", MemoryRemoteHandler) def _memory_factory(config: MemoryConfig, endpoint_name: str) -> MemoryDataSource: @@ -154,14 +153,11 @@ async def test_create_pipeline_from_config_supports_registered_custom_datasource await pipeline.close() -def test_build_config_supports_same_file_datasource_bootstrap(tmp_path: Path) -> None: +def test_build_config_supports_custom_datasource_type_loading(tmp_path: Path) -> None: register_datasource_type("memory_bootstrap", config_model=MemoryConfig, factory=_memory_factory, replace=True) profile_path = tmp_path / "custom-memory.yaml" profile_path.write_text( - "extensions:\n" - " bootstraps:\n" - " - memory_bootstrap_demo:register\n" "node_types: []\n" "local_datasource:\n" " type: memory_bootstrap\n" diff --git a/tests/unit/test_factory_endpoint_initialization.py b/tests/unit/test_factory_endpoint_initialization.py index 5c10ce7..17a5869 100644 --- a/tests/unit/test_factory_endpoint_initialization.py +++ b/tests/unit/test_factory_endpoint_initialization.py @@ -20,9 +20,9 @@ from sync_state_machine.pipeline.factory import create_pipeline_from_config from sync_state_machine.sync_system.strategy import DefaultSyncStrategy -class EndpointInitDbConfig(BaseDataSourceConfig): +class EndpointInitConfig(BaseDataSourceConfig): __test__ = False - type: str = "endpoint_init_db" + type: str = "endpoint_init_memory" class EndpointInitSchema(BaseModel): @@ -42,7 +42,7 @@ class EndpointInitStrategy(DefaultSyncStrategy[EndpointInitSchema]): schema = EndpointInitSchema -class EndpointInitDbHandler(BaseNodeHandler[EndpointInitSchema]): +class EndpointInitHandler(BaseNodeHandler[EndpointInitSchema]): __test__ = False def __init__(self, datasource: BaseDataSource): @@ -69,8 +69,7 @@ def _ensure_registered() -> None: reg.schema = EndpointInitSchema reg.node_class = EndpointInitNode reg.strategy_class = EndpointInitStrategy - reg.db_handler_class = EndpointInitDbHandler - reg.handler_classes["endpoint_init_db"] = EndpointInitDbHandler + reg.handler_classes["endpoint_init_memory"] = EndpointInitHandler return DomainRegistry.register( @@ -78,15 +77,14 @@ def _ensure_registered() -> None: schema=EndpointInitSchema, node_class=EndpointInitNode, strategy_class=EndpointInitStrategy, - db_handler_class=EndpointInitDbHandler, ) - DomainRegistry.register_handler(node_type, "endpoint_init_db", EndpointInitDbHandler) + DomainRegistry.register_handler(node_type, "endpoint_init_memory", EndpointInitHandler) def _register_endpoint_init_datasource_type() -> None: register_datasource_type( - "endpoint_init_db", - config_model=EndpointInitDbConfig, + "endpoint_init_memory", + config_model=EndpointInitConfig, factory=lambda config, endpoint_name: BaseDataSource(), replace=True, ) @@ -102,13 +100,13 @@ async def test_factory_initializes_same_datasource_type_with_distinct_handler_co config = PipelineRunConfig( node_types=["factory_endpoint_demo"], - local_datasource=EndpointInitDbConfig( - type="endpoint_init_db", + local_datasource=EndpointInitConfig( + type="endpoint_init_memory", handler_configs={"factory_endpoint_demo": {"side": "local", "batch_size": 10}}, target_project_ids=["project_local"], ), - remote_datasource=EndpointInitDbConfig( - type="endpoint_init_db", + remote_datasource=EndpointInitConfig( + type="endpoint_init_memory", handler_configs={"factory_endpoint_demo": {"side": "remote", "batch_size": 20}}, target_project_ids=["project_remote"], ), @@ -122,8 +120,8 @@ async def test_factory_initializes_same_datasource_type_with_distinct_handler_co local_handler = pipeline.local_datasource.get_handler("factory_endpoint_demo") remote_handler = pipeline.remote_datasource.get_handler("factory_endpoint_demo") - assert isinstance(local_handler, EndpointInitDbHandler) - assert isinstance(remote_handler, EndpointInitDbHandler) + assert isinstance(local_handler, EndpointInitHandler) + assert isinstance(remote_handler, EndpointInitHandler) assert local_handler.handler_config == {"side": "local", "batch_size": 10} assert remote_handler.handler_config == {"side": "remote", "batch_size": 20} finally: diff --git a/tests/unit/test_registry_unified_handler_lookup.py b/tests/unit/test_registry_unified_handler_lookup.py index 9266517..8dd6fa8 100644 --- a/tests/unit/test_registry_unified_handler_lookup.py +++ b/tests/unit/test_registry_unified_handler_lookup.py @@ -22,7 +22,7 @@ class RegistryLookupApiHandler: __test__ = False -class RegistryLookupDbHandler: +class RegistryLookupCustomHandler: __test__ = False @@ -33,7 +33,7 @@ def test_registry_supports_unified_handler_lookup() -> None: reg = DomainRegistry.get_registration(node_type) reg.jsonl_handler_class = RegistryLookupJsonlHandler reg.api_handler_class = RegistryLookupApiHandler - reg.db_handler_class = RegistryLookupDbHandler + reg.handler_classes["custom"] = RegistryLookupCustomHandler else: DomainRegistry.register( node_type=node_type, @@ -41,9 +41,9 @@ def test_registry_supports_unified_handler_lookup() -> None: node_class=RegistryLookupNode, jsonl_handler_class=RegistryLookupJsonlHandler, api_handler_class=RegistryLookupApiHandler, - db_handler_class=RegistryLookupDbHandler, ) + DomainRegistry.register_handler(node_type, "custom", RegistryLookupCustomHandler) assert DomainRegistry.get_handler(node_type, "jsonl") is RegistryLookupJsonlHandler assert DomainRegistry.get_handler(node_type, "api") is RegistryLookupApiHandler - assert DomainRegistry.get_handler(node_type, "db") is RegistryLookupDbHandler + assert DomainRegistry.get_handler(node_type, "custom") is RegistryLookupCustomHandler diff --git a/uv.lock b/uv.lock index 696e12c..c3126cf 100644 --- a/uv.lock +++ b/uv.lock @@ -122,7 +122,7 @@ wheels = [ [[package]] name = "ecm-sync-system" version = "0.1.0" -source = { virtual = "." } +source = { editable = "." } dependencies = [ { name = "aiosqlite" }, { name = "email-validator" },