ecm_sync_system/docs/library_integration_guide.md

sync_state_machine 接入手册

1. 目标

本项目现在可以作为可编辑安装库使用：

• 源码开发：pip install -e .
• 代码引用：import sync_state_machine
• 命令行运行：ecm-sync-run --config_path=...

推荐接入方式：

• sync_state_machine 负责同步引擎、状态机、pipeline、datasource 抽象。
• 业务系统负责提供自己的 datasource / handler 适配层，不要反向让 sync_state_machine 依赖业务系统。

---

2. 安装方式

2.1 本仓库本地安装

在仓库根目录执行：

pip install -e .

安装后可直接使用：

import sync_state_machine
print(sync_state_machine.__version__)

也可以直接走命令：

ecm-sync-run --config_path=run_profiles/preset/jsonl_to_api.default.yaml

---

3. 对外可直接引用的入口

常用入口：

• sync_state_machine.build_config
• sync_state_machine.build_default_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 注册会在导入包时自动完成。

---

4. 最小接入示例

4.1 直接运行现有 pipeline

from pathlib import Path

from sync_state_machine import build_config, load_overrides_from_file, run_pipeline_from_config


async def main() -> None:
    project_root = Path("/path/to/ecm_sync_system")
    profile_path = project_root / "run_profiles" / "preset" / "jsonl_to_api.default.yaml"

    overrides = load_overrides_from_file(profile_path, project_root)
    config = build_config(project_root=project_root, overrides=overrides)
    await run_pipeline_from_config(config, title="demo", print_summary=True)

适合场景：

• 仍沿用当前仓库的 domain 注册
• 仍沿用当前仓库的 JSONL / API handler
• 外部系统只想“调用引擎”

---

5. 数据源怎么引用

当前引擎的 datasource 分两层：

1. DataSource
   • 管理加载、批量执行、轮询、collection 注入、handler 注册。
2. Handler
   • 只处理某一个 node_type 的 I/O 逻辑。

因此接入时，不是“直接把业务 ORM 塞进 pipeline”，而是：

• 先确定你的数据源类型
• 再决定复用已有 datasource，还是只新增 handler

5.1 已有两类 datasource

JSONL

可直接复用：

• sync_state_machine.datasource.JsonlDataSource
• sync_state_machine.datasource.BaseJsonlHandler

适合：

• 本地 fixture
• 离线回放
• 调试数据集

API

可直接复用：

• sync_state_machine.datasource.ApiDataSource
• sync_state_machine.datasource.ApiClient
• sync_state_machine.datasource.BaseApiHandler

适合：

• HTTP 接口推送
• 异步 push + poll
• 远端系统同步

---

6. 怎么适配数据源

6.1 最推荐：复用已有 datasource，只写 handler

这是最小成本方案。

比如你的本地数据来自 backend 的数据库，但你不想改引擎核心，可以：

• 保持 ApiDataSource / JsonlDataSource 不变
• 新增你自己的 handler
• 在 handler 里把 backend 数据模型转换成节点 schema

如果是 HTTP 源：继承 BaseApiHandler。

如果是 JSONL 源：继承 BaseJsonlHandler。

API handler 示例

from pydantic import BaseModel

from sync_state_machine.datasource.api.handler import BaseApiHandler


class DemoSchema(BaseModel):
    id: str
    name: str


class DemoApiHandler(BaseApiHandler[DemoSchema]):
    _node_type = "demo"
    _node_class = DemoSyncNode
    _schema = DemoSchema

    async def load(self):
        payload = await self.api_client.get("/demo/list")
        return [self._create_node(item) for item in payload["data"]]

    async def create_all(self, nodes):
        ...

    async def update_all(self, nodes):
        ...

    async def delete_all(self, nodes):
        ...

    async def poll_tasks(self, task_ids):
        ...

JSONL handler 示例

from sync_state_machine.datasource.jsonl.handler import BaseJsonlHandler


class DemoJsonlHandler(BaseJsonlHandler):
    def __init__(self, datasource):
        super().__init__(
            datasource=datasource,
            node_type="demo",
            node_class=DemoSyncNode,
            schema=DemoSchema,
        )

然后把它注册到 DomainRegistry。

---

6.2 如果 backend 是数据库：建议写 handler / adapter，不要让引擎直接依赖 ORM

推荐结构：

backend/
  app/
    integrations/
      ecm_sync/
        datasource/
          user_db_handler.py
          contract_db_handler.py
        mapper/
          user_mapper.py
          contract_mapper.py
        services/
          sync_runner.py

职责建议：

• mapper/
  • backend model -> sync schema
  • sync schema -> backend push payload
• datasource/handler
  • 调 repository / service 获取数据
  • 转成 SyncNode
• services/sync_runner.py
  • 组装配置并调用 pipeline

不要这样做：

• 不要在 sync_state_machine 包里 import backend 的 model/repository
• 不要让 DomainRegistry 注册逻辑散落在 backend 业务层之外
• 不要在 datasource 之外做额外的 project 过滤；项目范围应通过 handler_configs.project.data_id_filter 下沉到具体 handler 解释

---

6.3 什么时候需要新增 DataSource 类

只有当你的 I/O 模型与当前两种 datasource 都不匹配时，才新增 BaseDataSource 子类。

例如：

• MQ / Kafka 批量消费型源
• 直接数据库快照源
• 特殊 RPC 源

此时建议：

1. 继承 BaseDataSource
2. 保持接口与现有 datasource 一致：
   • register_handler()
   • set_collection()
   • load_all()
   • sync_all()
3. 把“节点类型差异”仍下沉到 handler，不要写死在 datasource 里

---

7. Domain 怎么注册

引擎通过 DomainRegistry 查找：

• schema
• node_class
• strategy_class
• jsonl_handler_class
• api_handler_class
• 以及任意通过 register_handler() 追加的 datasource handler

示例：

from sync_state_machine.common.registry import DomainRegistry

DomainRegistry.register(
    node_type="demo",
    schema=DemoSchema,
    node_class=DemoSyncNode,
    strategy_class=DemoStrategy,
    jsonl_handler_class=DemoJsonlHandler,
    api_handler_class=DemoApiHandler,
)

DomainRegistry.register_handler("demo", "custom_remote", DemoCustomHandler)

如果是 backend 接入，建议在启动入口集中注册，而不是分散在 controller / service 里。

---

8. backend 里的推荐接法

建议 backend 只新增一层 integration：

from pathlib import Path

from sync_state_machine import build_config, create_pipeline_from_config


async def run_demo_sync() -> None:
    project_root = Path("/path/to/ecm_sync_system")
    overrides = {
        "node_types": ["project", "contract"],
        "local_datasource": {
            "type": "jsonl",
            "jsonl_dir": str(project_root / "tests" / "fixtures" / "demo"),
        },
        "remote_datasource": {
            "type": "api",
            "api_base_url": "https://example.com",
            "api_uid": "xxx",
            "api_secret": "xxx",
          "handler_configs": {
            "project": {"data_id_filter": ["demo-project"]}
          },
        },
    }

    config = build_config(project_root=project_root, overrides=overrides)
    pipeline = await create_pipeline_from_config(config)
    await pipeline.run()

  如果 backend 自己提供本地数据库数据，不要把 backend model 直接灌进 pipeline；先经过 handler / mapper 转成 schema 再进入引擎。

---

9. 接入时的边界约束

建议保持下面的依赖方向：

• backend integration -> sync_state_machine
• sync_state_machine -X-> backend

这样做的好处：

• 引擎能独立测试
• backend 改 repository 不会直接污染引擎
• 后续可以单独发布库

---

10. 当前阶段建议

如果项目还在高频迭代，推荐顺序：

1. 先 pip install -e .
2. 先在 backend 做 integration 目录，不急着发私有包
3. 先接 1~2 个 node_type，跑通完整链路
4. 稳定后再考虑拆成正式发布包

---

11. 相关文件

• 包入口：sync_state_machine/__init__.py
• CLI / 运行入口：sync_state_machine/cli.py
• 配置工厂：sync_state_machine/config
• datasource 基础层：sync_state_machine/datasource
• pipeline 工厂：sync_state_machine/pipeline/factory.py
• domain 注册：sync_state_machine/common/registry.py
• backend 集成入口：backend/app/thirdparty/ecm_sync/services/sync_runner.py