From 4ce4004cdb34b2657cca4eb5f48373ddf13ffffb Mon Sep 17 00:00:00 2001 From: strepsiades Date: Mon, 9 Mar 2026 16:47:25 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E5=85=B6=E4=BB=96=E6=96=87?= =?UTF-8?q?=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 6 +- run_profiles/README.md | 5 +- run_profiles/preset/README.md | 5 +- sync_state_machine/datasource/README.md | 335 +++++++++--------------- tests/README.md | 2 +- 5 files changed, 136 insertions(+), 217 deletions(-) diff --git a/README.md b/README.md index 6880451..7096f20 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,6 @@ -# test_push_system +# ecm-sync-system + +可作为命令行工具运行，也可作为 `pip install -e .` 的本地可编辑安装库接入其他项目。 ## 1. 项目简介 @@ -50,7 +52,7 @@ 可参考： - `docs/library_integration_guide.md`（库化安装、datasource 引用方式、backend 适配方式） -- `sync_state_machine/datasource/README.md` +- `sync_state_machine/datasource/README.md`（datasource/handler 分层说明） - `docs/runtime_config_guide.md` - `tests/README.md` diff --git a/run_profiles/README.md b/run_profiles/README.md index d7a8f3d..813b744 100644 --- a/run_profiles/README.md +++ b/run_profiles/README.md @@ -12,7 +12,10 @@ - `run_profiles/jsonl_to_jsonl.local.yaml` - `run_profiles/jsonl_to_api.local.yaml` 2. 按需修改后运行： - - `python run.py --config_path=run_profiles/jsonl_to_jsonl.local.yaml` + - `ecm-sync-run --config_path=run_profiles/jsonl_to_jsonl.local.yaml` + +兼容旧入口： +- `python run.py --config_path=run_profiles/jsonl_to_jsonl.local.yaml` 支持占位符： - `${PROJECT_ROOT}`：运行时替换为项目根目录绝对路径。 diff --git a/run_profiles/preset/README.md b/run_profiles/preset/README.md index d38e4cf..15611af 100644 --- a/run_profiles/preset/README.md +++ b/run_profiles/preset/README.md @@ -9,7 +9,10 @@ 使用方式： 1. 复制模板到项目根目录 `run_profiles/`，并重命名为你自己的文件（如 `*.local.yaml`）。 2. 修改后通过入口运行： - - `python run.py --config_path=run_profiles/jsonl_to_jsonl.local.yaml` + - `ecm-sync-run --config_path=run_profiles/jsonl_to_jsonl.local.yaml` + +兼容旧入口： +- `python run.py --config_path=run_profiles/jsonl_to_jsonl.local.yaml` 说明： - `run_profiles/*.yaml` / `run_profiles/*.yml` 默认被 `.gitignore` 忽略，用于本地私有配置。 diff --git a/sync_state_machine/datasource/README.md b/sync_state_machine/datasource/README.md index 3c44b47..b8ff6db 100644 --- a/sync_state_machine/datasource/README.md +++ b/sync_state_machine/datasource/README.md @@ -1,252 +1,163 @@ -# DataSource Layer Implementation +# DataSource 接入说明 -## Overview +本文档说明当前 `sync_state_machine.datasource` 层的职责边界，以及新增 datasource / handler 时应如何接入。 -This document describes the new datasource layer for `sync_system_new`, which is completely decoupled from the sync strategy and state machine logic. +如需看完整库化接入方式，优先参考 [docs/library_integration_guide.md](../../docs/library_integration_guide.md)。 -## Architecture +## 1. 分层模型 -The datasource layer follows a pure execution model: +当前 datasource 层分为两部分： -1. **Load**: Reads data from backend and creates `SyncNode` instances -2. **Save**: Executes actions (CREATE/UPDATE/DELETE) based on `node.action` -3. **Update Status**: Sets `node.status` to SUCCESS or FAILED after execution +1. `DataSource` + - 管理 handler 注册 + - 管理 collection 注入 + - 负责按 `node_type` 组织 `load_all()` / `sync_all()` + - 汇总并回写 `TaskResult` -### Key Principles +2. `Handler` + - 负责某一个 `node_type` 的具体 I/O + - 负责把原始数据转成 `SyncNode` + - 负责执行 create / update / delete / poll -- **No Strategy Logic**: Datasource doesn't make decisions about what to sync -- **State-Driven**: Only reads `node.action` and `node.status` to determine behavior -- **No ID Resolution**: IDs are already resolved by the strategy layer -- **Backend Agnostic**: Abstract base class supports multiple backends +设计原则： -## Components +- datasource 不承载业务策略 +- strategy 不直接做底层 I/O +- 节点类型差异下沉到 handler -### 1. Base Protocol (`base.py`) +--- -#### `DataSourceProtocol` +## 2. 当前可直接复用的 datasource -Protocol defining the interface for all datasource implementations: +### 2.1 JSONL + +- `JsonlDataSource` +- `BaseJsonlHandler` + +适合： + +- fixture +- 本地回放 +- 离线调试 + +### 2.2 API + +- `ApiDataSource` +- `ApiClient` +- `BaseApiHandler` + +适合： + +- HTTP 推送 +- 异步任务轮询 +- 远端系统对接 + +--- + +## 3. 最推荐的扩展方式：只新增 handler + +多数情况下，不需要新增新的 datasource 类。 + +推荐做法： + +1. 复用现有 `JsonlDataSource` 或 `ApiDataSource` +2. 为你的 `node_type` 新增 handler +3. 在 handler 中完成字段映射、加载、写入和轮询 +4. 将 handler 注册到 `DomainRegistry` + +这样可以保持： + +- pipeline 不变 +- datasource 生命周期不变 +- 策略层不感知底层来源 + +--- + +## 4. handler 需要实现什么 + +### 4.1 JSONL handler + +通常继承 `BaseJsonlHandler`： ```python -async def load_nodes( - node_type: str, - collection: DataCollection, - **filters: Any -) -> None: - """Load data from backend and create SyncNodes""" - -async def save_nodes( - node_type: str, - collection: DataCollection -) -> None: - """Execute pending actions for nodes""" +class DemoJsonlHandler(BaseJsonlHandler): + def __init__(self, datasource): + super().__init__( + datasource=datasource, + node_type="demo", + node_class=DemoSyncNode, + schema=DemoSchema, + ) ``` -#### `BaseDataSource` +### 4.2 API handler -Abstract base class providing common functionality: +通常继承 `BaseApiHandler`，实现： -- `load_nodes`: Creates SyncNodes from raw data -- `save_nodes`: Executes actions and updates status -- `_handle_create/update/delete`: Action-specific handlers +- `load()` +- `create_all()` +- `update_all()` +- `delete_all()` +- `poll_tasks()` -Subclasses implement: -- `_load_raw_data`: Load from backend -- `_create_item`: Execute CREATE -- `_update_item`: Execute UPDATE -- `_delete_item`: Execute DELETE -- `_generate_id`: Generate new IDs +并按需要补充： -### 2. JSONL Implementation (`jsonl_datasource.py`) +- `extract_created_id()` +- `get_update_fields()` +- 依赖节点 context 读取逻辑 -#### `JsonlDataSource` +--- -JSONL-based datasource for local testing: +## 5. 什么时候新增 DataSource 类 -**Features:** -- Loads `.jsonl` files into memory -- Generates UUIDs for new items -- Supports filtering during load -- Supports read-only mode -- Persists changes back to files +只有在现有 JSONL / API 模型都不匹配时，才建议新增 `BaseDataSource` 子类。 -**File Structure:** -``` -data/ - ├── contract_1.jsonl - ├── project_1.jsonl - └── ... -``` +例如： -**Usage:** -```python -datasource = JsonlDataSource(Path("data/")) +- 数据来自数据库快照 +- 数据来自 MQ / Kafka +- 数据来自特殊 RPC 或 SDK -# Load nodes -await datasource.load_nodes("contract", collection, project_id="P1") +即便新增 datasource，也建议保留同样的职责边界： -# Execute actions -await datasource.save_nodes("contract", collection) +- datasource 负责调度 +- handler 负责节点类型差异 -# Persist to disk -await datasource.save_all() -``` +--- -### 3. Domain Handlers +## 6. 注册方式 -Domain-specific handlers provide business logic (optional): - -**Example: `domain/contract/jsonl_handler.py`** +接入新 `node_type` 时，最终仍通过 `DomainRegistry` 暴露给 pipeline： ```python -class ContractJsonlHandler: - @staticmethod - def validate_create(data: Any) -> None: - """Validate before create""" - - @staticmethod - def transform_for_create(data: Any) -> Dict[str, Any]: - """Transform data before create""" +DomainRegistry.register( + node_type="demo", + schema=DemoSchema, + node_class=DemoSyncNode, + strategy_class=DemoStrategy, + jsonl_handler_class=DemoJsonlHandler, + api_handler_class=DemoApiHandler, +) ``` -## Node Lifecycle +--- -### Loading Phase +## 7. 推荐阅读顺序 -1. Datasource reads raw data from backend -2. Creates `SyncNode` for each item: - - `node_id`: Unique session ID (e.g., "contract_C1") - - `data_id`: Backend ID (e.g., "C1") - - `origin_data`: Raw data from backend - - `data`: Copy of origin_data (may be modified by strategy) - - `action`: NONE (initial) - - `status`: PENDING (initial) +1. [docs/library_integration_guide.md](../../docs/library_integration_guide.md) +2. `sync_state_machine/datasource/handler.py` +3. `sync_state_machine/datasource/api/handler.py` +4. `sync_state_machine/datasource/jsonl/handler.py` +5. `sync_state_machine/pipeline/factory.py` -### Execution Phase +--- -For each node with `status == PENDING` and `action != NONE`: +## 8. 结论 -1. **CREATE**: - - Generate new ID via `_generate_id` - - Call `_create_item` with new ID and data - - Set `node.data_id` to new ID - - Set `node.status` to SUCCESS +如果你在接入新系统，优先判断： -2. **UPDATE**: - - Call `_update_item` with existing `data_id` and data - - Set `node.status` to SUCCESS +- 是否可以直接复用 `JsonlDataSource` / `ApiDataSource` +- 是否只需要新增 handler +- 是否需要在业务系统侧做 mapper / adapter -3. **DELETE**: - - Call `_delete_item` with `data_id` - - Set `node.status` to SUCCESS - -4. **Error Handling**: - - On exception: Set `node.status` to FAILED - - Set `node.error` to exception message - -## Integration with Sync System - -The datasource is used by the pipeline layer: - -```python -# 1. Load data -local_ds = JsonlDataSource(Path("local_data/")) -remote_ds = JsonlDataSource(Path("remote_data/")) - -local_collection = DataCollection("local") -remote_collection = DataCollection("remote") - -await local_ds.load_nodes("contract", local_collection) -await remote_ds.load_nodes("contract", remote_collection) - -# 2. Strategy layer determines actions -strategy = ContractSyncStrategy(...) -await strategy.bind(...) # Sets node.action - -# 3. Execute actions -await remote_ds.save_nodes("contract", remote_collection) -await remote_ds.save_all() -``` - -## Testing - -Comprehensive test suite in `tests/test_datasource.py`: - -- Basic loading and node creation -- Filtering during load -- CREATE/UPDATE/DELETE actions -- Error handling and status updates -- Read-only mode -- Multiple node types - -Run tests: -```bash -python -m pytest sync_system_new/tests/test_datasource.py -v -``` - -## Future Extensions - -### API DataSource - -For production use, implement `ApiDataSource`: - -```python -class ApiDataSource(BaseDataSource): - async def _load_raw_data(self, node_type: str, **filters): - # Call REST API - - async def _create_item(self, node_type: str, item_id: str, data): - # POST request - - async def _generate_id(self, node_type: str, data): - # Return None - server generates ID -``` - -### Database DataSource - -For SQL databases: - -```python -class DatabaseDataSource(BaseDataSource): - async def _load_raw_data(self, node_type: str, **filters): - # SELECT query - - async def _create_item(self, node_type: str, item_id: str, data): - # INSERT query -``` - -## Comparison with Legacy System - -### Legacy System (`sync_system/datasource/`) - -- Tightly coupled with RepositoryProtocol -- Mixed concerns (data access + business logic) -- Schema validation in datasource layer -- Different interface for each entity type - -### New System (`sync_system_new/datasource/`) - -- Decoupled from strategy and state machine -- Pure data access (business logic in domain handlers) -- Generic interface for all entity types -- Direct integration with SyncNode and DataCollection - -## Key Differences - -| Aspect | Legacy | New | -|--------|--------|-----| -| Coupling | Tight with sync policies | Decoupled from sync logic | -| Interface | Repository per entity | Generic load/save | -| ID Handling | Mixed responsibilities | IDs already resolved | -| State Management | Mixed with data access | SyncNode-based | -| Testing | Complex setup | Simple, focused tests | - -## Conclusion - -The new datasource layer provides: - -1. **Clear Separation**: Data access is isolated from sync logic -2. **Flexibility**: Easy to add new backend types -3. **Testability**: Simple, focused tests -4. **Type Safety**: Full integration with Pydantic models -5. **Maintainability**: Single responsibility per component +通常情况下，**先写 handler，而不是先写新的 datasource**。 diff --git a/tests/README.md b/tests/README.md index 53479dd..9238cd7 100644 --- a/tests/README.md +++ b/tests/README.md @@ -25,5 +25,5 @@ - 覆盖核心业务链路（bind/create/update 与 full pipeline）。 ## 大规模自动回归 -- 按轮迭代的大规模自动测试方法见 [docs/auto_test_spec/agent_auto_test_method.md](docs/auto_test_spec/agent_auto_test_method.md)。 +- 按轮迭代的大规模自动测试方法见 [auto_test_spec/agent_auto_test_method.md](auto_test_spec/agent_auto_test_method.md)。 - 执行时优先采用“单 domain 复现 -> 修复/分类 -> 全量聚合回归 -> 记录总计变化”的节奏，避免一次携带过多上下文。