@@ -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 ex ecu tion
1. `DataSource`
- 管理 handler 注册
- 管理 coll ection 注入
- 负责按 `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`)
---
#### `D ataS ourceProtocol`
## 2. 当前可直接复用的 d atas ource
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 d atas ource for local testing:
## 5. 什么时候新增 D ataS ource 类
**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 `S yncNode` 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. `s ync_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**。
strepsiades