# 数据源指南(DataSource Guide) 本文档记录当前项目中不同数据源的职责、能力与配置方式,并按数据源分类说明。 ## 1. 总览 当前实现包含两类数据源: - JSONL 数据源:面向本地文件读写 - API 数据源:面向远端 HTTP 接口推送与轮询 两者都基于统一编排层 `BaseDataSource`,具备一致的流程骨架: 1. 注册 Handler(按节点类型) 2. load 阶段加载节点 3. sync 阶段按动作执行 create/update/delete 4. 统一处理 TaskResult 与状态机写回 5. 统一异步任务轮询(IN_PROGRESS -> SUCCESS/FAILED) 这意味着: - 业务策略可复用,不需关心底层介质差异 - 数据源差异主要落在 Handler 与传输/存储细节 --- ## 2. JSONL 数据源 ### 2.1 主要功能 JSONL 数据源对应实现: - `sync_state_machine/datasource/jsonl/datasource.py` - `sync_state_machine/datasource/jsonl/handler.py` 核心能力: - 从目录中按 `{node_type}_N.jsonl` 规则发现并加载数据 - 批量 create/update/delete 时,先改内存缓存,再统一刷盘 - 支持 `read_only`,在只读模式下禁止写回 - 支持按项目过滤(`target_project_ids`)在 load 阶段预过滤 ### 2.2 行为特点 - load:逐行解析 JSONL,单行坏数据仅警告并继续 - create/update/delete:返回统一 `TaskResult`,由 DataSource 层统一更新状态 - 写回:默认写到 `{node_type}_1.jsonl` ### 2.3 关键配置 `JsonlDataSourceConfig` 字段: - `type: jsonl` - `jsonl_dir`: 数据目录 - `read_only`: 是否只读 - `target_project_ids`: 项目过滤列表 - `handler_configs`: 按节点类型的 handler 参数 --- ## 3. API 数据源 ### 3.1 主要功能 API 数据源对应实现: - `sync_state_machine/datasource/api/datasource.py` - `sync_state_machine/datasource/api/client.py` - `sync_state_machine/datasource/api/handler.py` 核心能力: - 统一 HTTP 请求入口(签名、uid 注入、错误日志、trace) - 统一 push 任务轮询(`/push/result`) - 统一请求链路追踪(submit/poll/load) - 统一限流(全局窗口限流) ### 3.2 API 全局限流(重点) 当前已支持全局异步限流,入口在 `ApiClient.request()` 前。 配置字段: - `api_rate_limit_max_requests`:窗口内最大请求数 - `api_rate_limit_window_seconds`:窗口秒数 默认值: - `api_rate_limit_max_requests: 10` - `api_rate_limit_window_seconds: 10.0` 含义: - 任意 10 秒窗口内最多发 10 个请求 - 超出后,后续请求在客户端侧等待,不直接打到服务端 实现机制(简述): - 全局 `asyncio.Lock` + 时间戳队列 - 发送前清理窗口外时间戳 - 若窗口未满,立即占位并发送 - 若窗口已满,按最早时间戳计算等待时间后重试 禁用方式: - `api_rate_limit_max_requests <= 0` 或 `api_rate_limit_window_seconds <= 0` ### 3.3 从 collection 按需提取数据(重点) 在 API 模式下,Handler 不是盲目请求,而是从 `collection` 按依赖提取后再调用 API。 常见模式: 1. load 前置依赖提取 - 例如合同 Handler 先从 collection 取 project 节点,再按 project_id 拉合同列表 2. create/update 提取上下文 - 从 `node.context` 或 `node.get_data()` 中提取依赖字段(如 `project_id`) 3. 节点复用/更新 - `_create_basic_node` 会优先按 `data_id` 命中已有节点,避免重复节点 好处: - 减少无效 API 调用(按依赖、按项目有选择地加载) - 保持节点关系与上下文完整(project/contract 等依赖链) ### 3.4 API 校验链路(重点) API 数据源中,校验分为三层: 1. 领域 schema 校验(请求前/响应后) - 例如 `ContractCreate.model_validate(...)`、`ContractUpdate.model_validate(...)` - load 响应也会在解析后转为领域 schema 2. 通用更新过滤与差异识别 - `_filter_update_data`:仅在 schema 相关字段有变化时提交更新 - `_get_schema_diff`:用于差异字段识别与调试输出 3. 轮询接口输入校验 - `PushIdsSchema` 校验 push_ids 格式后再请求 `/push/result` 设计原则: - 校验失败直接显式失败,不静默降级 - 保持状态机与错误日志可追踪 ### 3.5 轮询与结果回写 - create 可返回 IN_PROGRESS(异步)或 SUCCESS(同步) - `BaseDataSource` 统一轮询并写回节点状态 - 结果通过状态机事件落到 SUCCESS/FAILED - create 成功后会进行 `data_id` 回写与 collection 更新(必要时) --- ## 4. 配置说明(按数据源区分) ### 4.1 JSONL 数据源配置示例 ```yaml local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered read_only: false target_project_ids: [] handler_configs: {} ``` ### 4.2 API 数据源配置示例(含限流) ```yaml remote_datasource: type: api api_base_url: http://localhost:8000/api/third/v2 api_uid: your_uid api_secret: your_secret api_debug: true # 轮询配置 poll_max_retries: 1 poll_interval: 2.0 # 全局限流:10 秒最多 10 次请求 api_rate_limit_max_requests: 10 api_rate_limit_window_seconds: 10.0 target_project_ids: - "project-id-1" handler_configs: {} ``` ### 4.3 配置模型位置 - `sync_state_machine/config/datasource_config.py` - `JsonlDataSourceConfig` - `ApiDataSourceConfig` 配置是严格模式(`extra="forbid"`): - 未声明字段会报错 - 避免运行时拼写错误导致“配置看起来生效、实际未生效” --- ## 5. 运行建议与排障 ### 5.1 API 触发限流(429)时 优先调整: - 降低 `api_rate_limit_max_requests` - 提高 `api_rate_limit_window_seconds` 例如: - `api_rate_limit_max_requests: 8` - `api_rate_limit_window_seconds: 10.0` ### 5.2 轮询超时 可调整: - `poll_max_retries` - `poll_interval` 总等待时长约为: - `poll_max_retries * poll_interval` ### 5.3 校验失败 建议直接修数据或 schema,不建议绕过校验;当前链路会保留失败信息用于定位。 --- ## 6. 相关文件索引 - `sync_state_machine/datasource/datasource.py` - `sync_state_machine/datasource/jsonl/datasource.py` - `sync_state_machine/datasource/jsonl/handler.py` - `sync_state_machine/datasource/api/datasource.py` - `sync_state_machine/datasource/api/client.py` - `sync_state_machine/datasource/api/handler.py` - `sync_state_machine/config/datasource_config.py` - `run_profiles/jsonl_to_api.yaml` - `run_profiles/preset/jsonl_to_api.default.yaml`