Compare commits
2 Commits
f759984331
...
515c3d9388
| Author | SHA1 | Date | |
|---|---|---|---|
 strepsiades
|
515c3d9388 | ||
|
strepsiades
|
4ce4004cdb |
@@ -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`
|
||||
|
||||
|
||||
@@ -32,12 +32,12 @@
|
||||
|
||||
目前已经先实现了一版最小骨架：
|
||||
|
||||
- 测试总配置：[tests/fixtures/auto_sync/auto_test_config.yaml](tests/fixtures/auto_sync/auto_test_config.yaml)
|
||||
- 流水线测试配置：[run_profiles/jsonl_to_api.test.yaml](run_profiles/jsonl_to_api.test.yaml)
|
||||
- 初始化脚本：[scripts/auto_test_init.py](scripts/auto_test_init.py)
|
||||
- 统一运行脚本：[scripts/run_auto_test.py](scripts/run_auto_test.py)
|
||||
- 公共工具：[scripts/auto_test_lib.py](scripts/auto_test_lib.py)
|
||||
- 第一个 domain 用例：[tests/fixtures/auto_sync/domains/contract.yaml](tests/fixtures/auto_sync/domains/contract.yaml)
|
||||
- 测试总配置：[tests/fixtures/auto_sync/auto_test_config.yaml](../../tests/fixtures/auto_sync/auto_test_config.yaml)
|
||||
- 流水线测试配置：[run_profiles/jsonl_to_api.test.yaml](../../run_profiles/jsonl_to_api.test.yaml)
|
||||
- 初始化脚本：[scripts/auto_test_init.py](../../scripts/auto_test_init.py)
|
||||
- 统一运行脚本：[scripts/run_auto_test.py](../../scripts/run_auto_test.py)
|
||||
- 公共工具：[scripts/auto_test_lib.py](../../scripts/auto_test_lib.py)
|
||||
- 第一个 domain 用例：[tests/fixtures/auto_sync/domains/contract.yaml](../../tests/fixtures/auto_sync/domains/contract.yaml)
|
||||
|
||||
## 3. 配置管理
|
||||
|
||||
@@ -330,7 +330,7 @@
|
||||
- 一类后端读写不一致问题。
|
||||
|
||||
2. **先单 domain 复现**
|
||||
- 用 [scripts/run_auto_test.py](scripts/run_auto_test.py) 只跑目标 domain。
|
||||
- 用 [scripts/run_auto_test.py](../../scripts/run_auto_test.py) 只跑目标 domain。
|
||||
- 目的不是求“全绿”，而是先确认问题是否稳定复现。
|
||||
|
||||
3. **定位并修复一类根因**
|
||||
@@ -343,7 +343,7 @@
|
||||
- 记录这轮减少了多少失败 case。
|
||||
|
||||
5. **再跑聚合回归**
|
||||
- 用 [scripts/run_auto_test_all_domains.py](scripts/run_auto_test_all_domains.py) 执行全量聚合；
|
||||
- 用 [scripts/run_auto_test_all_domains.py](../../scripts/run_auto_test_all_domains.py) 执行全量聚合；
|
||||
- 当前聚合模式是“**每个 domain 独立初始化并单独执行**”，用于避免跨 domain 状态污染。
|
||||
|
||||
6. **输出轮次结论**
|
||||
|
||||
@@ -58,27 +58,27 @@ large_scale_test:
|
||||
## 使用方式
|
||||
|
||||
- 控制入口：调整本文档 frontmatter 中的 `rounds`
|
||||
- 执行脚本：[scripts/run_large_scale_auto_test.py](scripts/run_large_scale_auto_test.py)
|
||||
- 执行脚本：[scripts/run_large_scale_auto_test.py](../../scripts/run_large_scale_auto_test.py)
|
||||
- 默认命令：`python scripts/run_large_scale_auto_test.py`
|
||||
|
||||
## 设计约束
|
||||
|
||||
- 默认使用 `fresh` 初始化，优先保证每轮从干净环境开始。
|
||||
- `aggregate` runner 适合大规模基线回归；它内部会对每个 domain 独立初始化。
|
||||
- `domains` runner 适合小范围复核；它会按 [scripts/run_auto_test.py](scripts/run_auto_test.py) 的语义执行。
|
||||
- `domains` runner 适合小范围复核；它会按 [scripts/run_auto_test.py](../../scripts/run_auto_test.py) 的语义执行。
|
||||
- 失败不应靠 fallback 掩盖；应在轮次记录里明确分类。
|
||||
|
||||
## Runner 约定
|
||||
|
||||
### `aggregate`
|
||||
|
||||
- 调用 [scripts/run_auto_test_all_domains.py](scripts/run_auto_test_all_domains.py)
|
||||
- 调用 [scripts/run_auto_test_all_domains.py](../../scripts/run_auto_test_all_domains.py)
|
||||
- 适合建立全量基线
|
||||
- 最能避免跨 domain 污染
|
||||
|
||||
### `domains`
|
||||
|
||||
- 调用 [scripts/run_auto_test.py](scripts/run_auto_test.py)
|
||||
- 调用 [scripts/run_auto_test.py](../../scripts/run_auto_test.py)
|
||||
- 适合复核单个或少量 domain
|
||||
- 用于缩小上下文，验证某个问题是否稳定复现
|
||||
|
||||
|
||||
+2
-2
@@ -59,8 +59,8 @@
|
||||
|
||||
- `auto_bind=false`
|
||||
- 症状：MISSING 节点更容易进入 `WARNING`，不会走自动创建识别。
|
||||
- `skip_sync=true`
|
||||
- 症状：该类型只加载不执行 bind/create/update，常被误以为“没生效”。
|
||||
- `local_orphan_action=none` + `remote_orphan_action=none` + `update_direction=none`
|
||||
- 症状：该类型会被整体跳过，常被误以为“配置没生效”。
|
||||
- `load_mode=persisted_only`
|
||||
- 症状：数据只来自持久化，不主动拉取新数据。
|
||||
- `update_direction=none`
|
||||
|
||||
@@ -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}`：运行时替换为项目根目录绝对路径。
|
||||
|
||||
@@ -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` 忽略，用于本地私有配置。
|
||||
|
||||
@@ -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**。
|
||||
|
||||
+1
-1
@@ -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 复现 -> 修复/分类 -> 全量聚合回归 -> 记录总计变化”的节奏，避免一次携带过多上下文。
|
||||
|
||||
Reference in New Issue
Block a user