From 2c09c611650b296107d89159c65454dc2cfe7dc1 Mon Sep 17 00:00:00 2001 From: strepsiades Date: Fri, 20 Mar 2026 14:38:56 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E4=BA=86pipeline=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E9=A1=BA=E5=BA=8F=E3=80=82=E7=8E=B0=E5=9C=A8=E6=8C=89?= =?UTF-8?q?domain=E9=A1=BA=E5=BA=8F=E8=AF=BB=E5=8F=96=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=BA=90=E6=95=B0=E6=8D=AE=EF=BC=8C=E9=98=B2=E6=AD=A2=E5=89=8D?= =?UTF-8?q?=E9=9D=A2domain=E7=9A=84=E6=9B=B4=E6=96=B0=E5=BD=B1=E5=93=8D?= =?UTF-8?q?=E5=90=8E=E9=9D=A2domain.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将target_project_ids替换为更通用的data_id_filter --- docs/datasource_guide.md | 16 ++-- docs/factory_registry_refactor_plan.md | 1 - docs/library_integration_guide.md | 6 +- docs/runtime_config_guide.md | 9 +- docs/sync_flow.md | 10 +- run_profiles/preset/jsonl_to_api.default.yaml | 18 ++-- .../preset/jsonl_to_jsonl.default.yaml | 2 - scripts/auto_test_domains/base.py | 4 +- sync_state_machine/common/collection.py | 6 +- .../config/datasource_config.py | 2 - sync_state_machine/datasource/api/handler.py | 10 +- sync_state_machine/datasource/datasource.py | 19 ---- sync_state_machine/datasource/handler.py | 69 +++++++++++-- .../datasource/jsonl/handler.py | 13 +-- .../domain/preparation/api_handler.py | 9 -- .../domain/project/api_handler.py | 26 +---- sync_state_machine/pipeline/factory.py | 32 +++---- .../pipeline/full_sync_pipeline.py | 37 ++++--- tests/unit/test_auto_test_project_profile.py | 8 +- ...atasource_target_project_ids_resolution.py | 22 ++--- .../test_factory_db_handler_registration.py | 1 - .../test_factory_endpoint_initialization.py | 2 - ...st_full_sync_pipeline_downstream_reload.py | 96 +++++++++++++++++++ ...est_pipeline_strategy_continue_on_error.py | 5 + .../unit/test_project_api_handler_updates.py | 3 +- .../test_target_project_preload_filter.py | 10 +- ui/config_models.py | 3 - ui/runner.py | 2 - ui/web/index.html | 27 +++++- 29 files changed, 296 insertions(+), 172 deletions(-) create mode 100644 tests/unit/test_full_sync_pipeline_downstream_reload.py diff --git a/docs/datasource_guide.md b/docs/datasource_guide.md index 08e3935..16078a7 100644 --- a/docs/datasource_guide.md +++ b/docs/datasource_guide.md @@ -38,7 +38,7 @@ JSONL 数据源对应实现： - 从目录中按 `{node_type}_N.jsonl` 规则发现并加载数据 - 批量 create/update/delete 时，先改内存缓存，再统一刷盘 - 支持 `read_only`，在只读模式下禁止写回 -- 支持按项目过滤（`target_project_ids`）在 load 阶段预过滤 +- 支持按项目过滤（`handler_configs.project.data_id_filter`）在 load 阶段预过滤 ### 2.2 行为特点 @@ -53,7 +53,7 @@ JSONL 数据源对应实现： - `type: jsonl` - `jsonl_dir`: 数据目录 - `read_only`: 是否只读 -- `target_project_ids`: 项目过滤列表 +- `handler_configs.project.data_id_filter`: 项目过滤列表 - `handler_configs`: 按节点类型的 handler 参数 --- @@ -159,8 +159,9 @@ local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered read_only: false - target_project_ids: [] - handler_configs: {} + handler_configs: + project: + data_id_filter: [] ``` ### 4.2 API 数据源配置示例（含限流） @@ -181,9 +182,10 @@ remote_datasource: api_rate_limit_max_requests: 10 api_rate_limit_window_seconds: 10.0 - target_project_ids: - - "project-id-1" - handler_configs: {} + handler_configs: + project: + data_id_filter: + - "project-id-1" ``` ### 4.3 配置模型位置 diff --git a/docs/factory_registry_refactor_plan.md b/docs/factory_registry_refactor_plan.md index e88b10c..b558065 100644 --- a/docs/factory_registry_refactor_plan.md +++ b/docs/factory_registry_refactor_plan.md @@ -93,7 +93,6 @@ - endpoint 创建 handler - endpoint 统一注入： - `handler_config` - - `target_project_ids` - datasource-specific runtime object（如 api client） 最终让 `factory` 只负责： diff --git a/docs/library_integration_guide.md b/docs/library_integration_guide.md index 179f64f..1c93619 100644 --- a/docs/library_integration_guide.md +++ b/docs/library_integration_guide.md @@ -239,7 +239,7 @@ backend/ - 不要在 `sync_state_machine` 包里 import backend 的 model/repository - 不要让 `DomainRegistry` 注册逻辑散落在 backend 业务层之外 -- 不要在 datasource 之外做额外的 project 过滤；`target_project_ids` 只应由 datasource 自己解释 +- 不要在 datasource 之外做额外的 project 过滤；项目范围应通过 `handler_configs.project.data_id_filter` 下沉到具体 handler 解释 --- @@ -320,7 +320,9 @@ async def run_demo_sync() -> None: "api_base_url": "https://example.com", "api_uid": "xxx", "api_secret": "xxx", - "target_project_ids": ["demo-project"], + "handler_configs": { + "project": {"data_id_filter": ["demo-project"]} + }, }, } diff --git a/docs/runtime_config_guide.md b/docs/runtime_config_guide.md index bde57c8..5271a4e 100644 --- a/docs/runtime_config_guide.md +++ b/docs/runtime_config_guide.md @@ -18,9 +18,6 @@ - `node_types` - 本轮参与同步的业务类型列表。 - 顺序会影响依赖链执行时机（当前按 node_type 串行执行 bind/create/update）。 -- `target_project_ids` - - 兼容字段。作为两侧 datasource 的项目白名单兜底值。 - - 若 datasource 自身配置了 `target_project_ids`，则优先使用 datasource 级配置。 - `local_datasource` / `remote_datasource` - 分别定义两侧数据来源与执行端。 - `strategies` @@ -37,8 +34,8 @@ - `type=jsonl`：使用本地 JSONL 文件作为数据源。 - `jsonl_dir`：JSONL 目录路径。 - `read_only=false`：允许该侧执行 create/update 回写；若为 true，通常用于只读加载。 -- `target_project_ids=[]`：该侧项目白名单；为空时不进行项目筛选。 -- `handler_configs`：预留给各业务 handler 的扩展参数。 +- `handler_configs.project.data_id_filter=[]`：项目白名单；为空时不进行项目筛选。 +- `handler_configs`：按节点类型下发给各业务 handler 的扩展参数。 ### 3.2 API 数据源 @@ -46,7 +43,7 @@ - `api_base_url` / `api_uid` / `api_secret`：接口访问参数。 - `api_debug`：开启后输出更多调试信息。 - `poll_max_retries` / `poll_interval`：异步任务轮询次数与间隔。 -- `target_project_ids`：**必填**，该侧项目白名单；至少需要指定一个项目ID，以避免加载过多远程数据。 +- `handler_configs.project.data_id_filter`：建议显式配置的项目白名单；至少需要指定一个项目ID，以避免加载过多远程数据。 - `handler_configs`：业务 handler 扩展参数。 ## 4. 策略字段（Strategy） diff --git a/docs/sync_flow.md b/docs/sync_flow.md index 0b12361..626f8fa 100644 --- a/docs/sync_flow.md +++ b/docs/sync_flow.md @@ -2,10 +2,12 @@ ## 1. 全流程主线 1. pipeline 装配组件并加载持久化。 -2. pipeline 调用 `strategy.bind()`。 -3. pipeline 调用 `strategy.create()` / `strategy.update()` 生成动作节点。 -4. pipeline 调用 `datasource.sync_all()` 执行动作并回写执行状态。 -5. pipeline 输出统计并持久化。 +2. pipeline 为 datasource 绑定 collection，但不在 bootstrap 阶段全量拉取业务数据。 +3. pipeline 在每个 `node_type` 开始执行前，单独加载该 `node_type` 的最新数据。 +4. pipeline 调用 `strategy.bind()`。 +5. pipeline 调用 `strategy.create()` / `strategy.update()` 生成动作节点。 +6. pipeline 调用 `datasource.sync_all()` 执行动作并回写执行状态。 +7. pipeline 输出统计并持久化。 ## 2. reset 时序 diff --git a/run_profiles/preset/jsonl_to_api.default.yaml b/run_profiles/preset/jsonl_to_api.default.yaml index 0eba99c..45dd7a7 100644 --- a/run_profiles/preset/jsonl_to_api.default.yaml +++ b/run_profiles/preset/jsonl_to_api.default.yaml @@ -22,10 +22,11 @@ local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered read_only: false - target_project_ids: - - "0f9c3e22-f4bb-4803-825a-4f238eeb06d5" - - "f3e02e84-e81c-4aa6-88d4-f5aa108c8227" - handler_configs: {} + handler_configs: + project: + data_id_filter: + - "0f9c3e22-f4bb-4803-825a-4f238eeb06d5" + - "f3e02e84-e81c-4aa6-88d4-f5aa108c8227" remote_datasource: type: api api_base_url: http://localhost:8000/api/third/v2 @@ -36,11 +37,12 @@ remote_datasource: api_rate_limit_window_seconds: 10.0 poll_max_retries: 1 poll_interval: 0.5 - target_project_ids: - # 请替换为实际的项目ID，至少需要一个 - - "0f9c3e22-f4bb-4803-825a-4f238eeb06d5" - - "f3e02e84-e81c-4aa6-88d4-f5aa108c8227" handler_configs: + project: + data_id_filter: + # 请替换为实际的项目ID，至少需要一个 + - "0f9c3e22-f4bb-4803-825a-4f238eeb06d5" + - "f3e02e84-e81c-4aa6-88d4-f5aa108c8227" supplier: {} persist: backend: sqlite diff --git a/run_profiles/preset/jsonl_to_jsonl.default.yaml b/run_profiles/preset/jsonl_to_jsonl.default.yaml index dbfbf8f..74e1925 100644 --- a/run_profiles/preset/jsonl_to_jsonl.default.yaml +++ b/run_profiles/preset/jsonl_to_jsonl.default.yaml @@ -23,13 +23,11 @@ local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered read_only: false - target_project_ids: [] handler_configs: {} remote_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource read_only: false - target_project_ids: [] handler_configs: {} persist: backend: sqlite diff --git a/scripts/auto_test_domains/base.py b/scripts/auto_test_domains/base.py index 02271bc..fca3c5e 100644 --- a/scripts/auto_test_domains/base.py +++ b/scripts/auto_test_domains/base.py @@ -292,8 +292,8 @@ def build_temp_run_profile_for_project_auto_bind( project_detail_push_keys: set[str] | None = None, ) -> tuple[str, list[str]]: profile = load_yaml_config(config["pipeline"]["run_profile"]) - profile.setdefault("local_datasource", {})["target_project_ids"] = [local_project_id] - profile.setdefault("remote_datasource", {})["target_project_ids"] = [remote_project_id] + profile.setdefault("local_datasource", {}).setdefault("handler_configs", {}).setdefault("project", {})["data_id_filter"] = [local_project_id] + profile.setdefault("remote_datasource", {}).setdefault("handler_configs", {}).setdefault("project", {})["data_id_filter"] = [remote_project_id] project_strategy = profile.setdefault("strategies", {}).setdefault("project", {}).setdefault("config", {}) project_strategy["auto_bind"] = False diff --git a/sync_state_machine/common/collection.py b/sync_state_machine/common/collection.py index 70649ca..f9ba777 100644 --- a/sync_state_machine/common/collection.py +++ b/sync_state_machine/common/collection.py @@ -514,9 +514,9 @@ class DataCollection: await self.persistence.save_nodes_bulk(self.collection_id, list(self._nodes.values())) - async def filter_by_project_ids(self, target_project_ids: List[str]) -> int: - """按目标项目集合过滤当前 collection，返回删除数量。""" - target_set = set(target_project_ids or []) + async def filter_by_project_ids(self, data_id_filter: List[str]) -> int: + """按目标项目 data_id 集合过滤当前 collection，返回删除数量。""" + target_set = set(data_id_filter or []) if not target_set: return 0 diff --git a/sync_state_machine/config/datasource_config.py b/sync_state_machine/config/datasource_config.py index 4e2b01f..1c19829 100644 --- a/sync_state_machine/config/datasource_config.py +++ b/sync_state_machine/config/datasource_config.py @@ -11,7 +11,6 @@ class BaseDataSourceConfig(BaseModel): model_config = ConfigDict(extra="forbid", validate_assignment=True) type: str - target_project_ids: List[str] = Field(default_factory=list) handler_configs: Dict[str, Dict[str, Any]] = Field(default_factory=dict) @@ -33,7 +32,6 @@ class ApiDataSourceConfig(BaseDataSourceConfig): poll_interval: float = 0.5 api_rate_limit_max_requests: int = 10 api_rate_limit_window_seconds: float = 10.0 - target_project_ids: List[str] = Field(default_factory=list) DataSourceConfig = Annotated[ diff --git a/sync_state_machine/datasource/api/handler.py b/sync_state_machine/datasource/api/handler.py index 8abc697..3efe7e4 100644 --- a/sync_state_machine/datasource/api/handler.py +++ b/sync_state_machine/datasource/api/handler.py @@ -69,9 +69,13 @@ class BaseApiHandler(NodeHandler[T]): def set_handler_config(self, config: Dict[str, Any]) -> None: self.handler_config = dict(config) - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - """默认忽略项目过滤列表，具体子类按需覆盖。""" - return + def get_data_id_filter(self) -> List[str]: + value = self.handler_config.get("data_id_filter", []) + if isinstance(value, str): + return [value] if value else [] + if isinstance(value, list): + return [str(item) for item in value if str(item)] + return [] @property def node_type(self) -> str: diff --git a/sync_state_machine/datasource/datasource.py b/sync_state_machine/datasource/datasource.py index 61e2629..a29bbb3 100644 --- a/sync_state_machine/datasource/datasource.py +++ b/sync_state_machine/datasource/datasource.py @@ -74,7 +74,6 @@ class BaseDataSource(ABC): def __init__(self, poll_max_retries: int = 1, poll_interval: float = 0.5): self._handlers: Dict[str, "NodeHandler"] = {} self._collection: Optional["DataCollection"] = None - self._target_project_ids: List[str] = [] # 按节点类型统计信息 # {"company": {"loaded": 10, "synced": 10, "success": 10, "failed": 0}, ...} @@ -87,13 +86,6 @@ class BaseDataSource(ABC): self._poll_interval = max(0, poll_interval) self._sm_runtime: Optional[StateMachineRuntime] = None - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - self._target_project_ids = [str(item) for item in target_project_ids if str(item)] - - @property - def target_project_ids(self) -> List[str]: - return list(self._target_project_ids) - def _ensure_sm_runtime(self) -> StateMachineRuntime: if self._sm_runtime is None: cfg_path = Path(__file__).resolve().parents[1] / "config" / "node_state_machine.yaml" @@ -379,17 +371,6 @@ class BaseDataSource(ABC): datasource.register_handler(ContractNodeHandler(api_client)) """ self._handlers[handler.node_type] = handler - self._apply_target_project_ids_to_handler(handler) - - def set_target_project_ids(self, target_project_ids: Optional[List[str]]) -> None: - """设置项目过滤列表，并在注册阶段下发给所有 handler。""" - normalized = [str(pid) for pid in (target_project_ids or []) if str(pid)] - self._target_project_ids = normalized - for handler in self._handlers.values(): - self._apply_target_project_ids_to_handler(handler) - - def _apply_target_project_ids_to_handler(self, handler: "NodeHandler") -> None: - handler.set_target_project_ids(self._target_project_ids) def get_handler(self, node_type: str) -> "NodeHandler": """获取 Handler""" diff --git a/sync_state_machine/datasource/handler.py b/sync_state_machine/datasource/handler.py index e45539a..eebef21 100644 --- a/sync_state_machine/datasource/handler.py +++ b/sync_state_machine/datasource/handler.py @@ -74,10 +74,6 @@ class NodeHandler(Protocol[T]): """注入 handler 级别配置。默认可忽略。""" ... - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - """注入项目过滤列表。默认可忽略。""" - ... - def set_api_client(self, client: "ApiClient") -> None: """注入 API 客户端。默认可忽略。""" ... @@ -360,14 +356,18 @@ class BaseNodeHandler(ABC, Generic[T]): """ self._collection = collection - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - """默认忽略项目过滤能力。""" - return - def set_handler_config(self, config: Dict[str, Any]) -> None: """默认保存 handler 级别配置，供子类按需读取。""" self.handler_config = dict(config) + def get_data_id_filter(self) -> List[str]: + value = self.handler_config.get("data_id_filter", []) + if isinstance(value, str): + return [value] if value else [] + if isinstance(value, list): + return [str(item) for item in value if str(item)] + return [] + def set_api_client(self, client: "ApiClient") -> None: """默认忽略 API 客户端注入。""" return @@ -479,6 +479,59 @@ class BaseNodeHandler(ABC, Generic[T]): return self.create_node(data) +class CompatibleNodeHandler(BaseNodeHandler[T]): + def __init__(self, handler: BaseNodeHandler[T]): + super().__init__(handler.node_type, handler.node_class, handler.schema) + self._handler = handler + + def set_collection(self, collection: "DataCollection") -> None: + super().set_collection(collection) + self._handler.set_collection(collection) + + def set_handler_config(self, config: Dict[str, Any]) -> None: + self._handler.set_handler_config(config) + self.handler_config = dict(self._handler.handler_config) + + def set_api_client(self, client: "ApiClient") -> None: + self._handler.set_api_client(client) + + def set_poll_mode(self, mode: str) -> None: + self._handler.set_poll_mode(mode) + + def get_update_fields(self, *, exclude: frozenset[str] = frozenset({"id"})) -> List[str]: + return self._handler.get_update_fields(exclude=exclude) + + async def load(self) -> List['SyncNode']: + return await self._handler.load() + + async def create_all( + self, + nodes: List["SyncNode"] + ) -> List[TaskResult]: + return await self._handler.create_all(nodes) + + async def update_all( + self, + nodes: List["SyncNode"] + ) -> List[TaskResult]: + return await self._handler.update_all(nodes) + + async def delete_all( + self, + nodes: List["SyncNode"] + ) -> List[TaskResult]: + return await self._handler.delete_all(nodes) + + async def poll_tasks(self, task_ids: List[str]) -> Dict[str, TaskResult]: + return await self._handler.poll_tasks(task_ids) + + async def validate(self, data: Dict[str, Any]) -> ValidationResult: + return await self._handler.validate(data) + + def extract_id(self, data: Dict[str, Any]) -> str: + return self._handler.extract_id(data) + + class NodeHandlerRegistry: """节点处理器注册表""" diff --git a/sync_state_machine/datasource/jsonl/handler.py b/sync_state_machine/datasource/jsonl/handler.py index f26fc2e..a6f5f3f 100644 --- a/sync_state_machine/datasource/jsonl/handler.py +++ b/sync_state_machine/datasource/jsonl/handler.py @@ -50,22 +50,17 @@ class BaseJsonlHandler(BaseNodeHandler): ): super().__init__(node_type, node_class, schema) self.datasource = datasource - self._target_project_ids: List[str] = [] - - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - """注入项目过滤列表的本地副本。真正的过滤来源仍然是 datasource。""" - self._target_project_ids = [str(pid) for pid in target_project_ids if str(pid)] def _should_skip_item_by_project_filter(self, item: Dict[str, Any], item_id: str) -> bool: - target_project_ids = set(self.datasource.target_project_ids) - if not target_project_ids: + data_id_filter = set(self.get_data_id_filter()) + if not data_id_filter: return False if self.node_type == "project": - return item_id not in target_project_ids + return item_id not in data_id_filter project_id = item.get("project_id") if project_id: - return str(project_id) not in target_project_ids + return str(project_id) not in data_id_filter return False async def load(self) -> List['SyncNode']: diff --git a/sync_state_machine/domain/preparation/api_handler.py b/sync_state_machine/domain/preparation/api_handler.py index cc1c04b..0cd0bea 100644 --- a/sync_state_machine/domain/preparation/api_handler.py +++ b/sync_state_machine/domain/preparation/api_handler.py @@ -27,17 +27,8 @@ class PreparationApiHandler(BaseApiHandler): self._node_type = "preparation" self._node_class = PreparationSyncNode self._schema = PreparationDetail - self._filter_project_id = None self.update_schemas = [PostPreparation] - def set_filter_project_id(self, project_id: str | List[str]): - """设置要加载的项目ID过滤列表""" - self._filter_project_id = project_id - - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - """设置目标项目ID列表""" - self.set_filter_project_id(target_project_ids) - async def load(self) -> List[SyncNode]: """从 API 加载里程碑数据（需要项目上下文）""" from ..project.api_handler import ProjectApiHandler diff --git a/sync_state_machine/domain/project/api_handler.py b/sync_state_machine/domain/project/api_handler.py index 0e6a831..7c513de 100644 --- a/sync_state_machine/domain/project/api_handler.py +++ b/sync_state_machine/domain/project/api_handler.py @@ -71,26 +71,11 @@ class ProjectApiHandler(BaseApiHandler): # 类级别缓存：存储 all_info 数据 _all_info_cache: Dict[str, Dict[str, Any]] = {} - def __init__(self, filter_project_id: str | List[str] | None = None): - """ - 初始化 Project Handler - - Args: - filter_project_id: 指定只加载的项目ID列表（可选，用于多项目测试） - """ + def __init__(self): super().__init__() self._node_type = "project" self._node_class = ProjectSyncNode self._schema = ProjectResponseBase - if filter_project_id is None: - self._filter_project_id = None - elif isinstance(filter_project_id, str): - self._filter_project_id = [filter_project_id] - else: - self._filter_project_id = filter_project_id # 保存过滤条件 - - def set_target_project_ids(self, target_project_ids: List[str]) -> None: - self._filter_project_id = list(target_project_ids) if target_project_ids else None @classmethod async def get_all_info(cls, client, project_id: str) -> Dict[str, Any]: @@ -127,15 +112,12 @@ class ProjectApiHandler(BaseApiHandler): nodes = [] - # 确定要加载的项目ID列表：参数 > 构造函数配置 - target_project_ids = self._filter_project_id or [] - if isinstance(target_project_ids, str): - target_project_ids = [target_project_ids] + project_id_filter = self.get_data_id_filter() - if target_project_ids: + if project_id_filter: # 只加载指定项目列表 projects : List[ProjectResponseBase]= [] - for project_id in target_project_ids: + for project_id in project_id_filter: project_response = await api_get_project(self.api_client, project_id) projects.append(project_response) else: diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index 1000c88..57cb82a 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -277,9 +277,6 @@ async def create_pipeline_from_config( project_root=_project_root(), ) await _initialize_datasource(remote_ds, endpoint_name="remote") - local_ds.set_target_project_ids([str(pid) for pid in config.local_datasource.target_project_ids if str(pid)]) - remote_ds.set_target_project_ids([str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)]) - _register_handlers(config, local_ds, remote_ds, all_types) local_collection = DataCollection("local", persistence, auto_persist=False) @@ -434,8 +431,6 @@ async def create_jsonl_to_api_pipeline( api_secret: str, # 节点类型 node_types: List[str], - # 项目过滤 - target_project_ids: Optional[List[str]] = None, strategy_overrides: Optional[Dict[str, Dict[str, Any]]] = None, # Pipeline 行为 enable_persistence: bool = True, @@ -446,8 +441,8 @@ async def create_jsonl_to_api_pipeline( poll_interval: float = 0.5, api_rate_limit_max_requests: int = 30, api_rate_limit_window_seconds: float = 10.0, - # Handler 自定义配置（可选，作用于 API 端） - handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, + local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, + remote_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, # Logger 配置 initialize_logger: bool = True, logger_file_path: Optional[str] = None, @@ -459,7 +454,7 @@ async def create_jsonl_to_api_pipeline( type="jsonl", jsonl_dir=local_jsonl_dir, read_only=False, - target_project_ids=target_project_ids or [], + handler_configs=local_handler_configs or {}, ), remote_datasource=ApiDataSourceConfig( type="api", @@ -471,8 +466,7 @@ async def create_jsonl_to_api_pipeline( poll_interval=poll_interval, api_rate_limit_max_requests=api_rate_limit_max_requests, api_rate_limit_window_seconds=api_rate_limit_window_seconds, - target_project_ids=target_project_ids or [], - handler_configs=handler_configs or {}, + handler_configs=remote_handler_configs or {}, ), strategies=[], persist=PersistConfig( @@ -502,8 +496,6 @@ async def create_api_to_jsonl_pipeline( persistence_db: str, # 节点类型 node_types: List[str], - # 项目过滤 - target_project_ids: Optional[List[str]] = None, strategy_overrides: Optional[Dict[str, Dict[str, Any]]] = None, # Pipeline 行为 enable_persistence: bool = True, @@ -514,8 +506,8 @@ async def create_api_to_jsonl_pipeline( poll_interval: float = 0.5, api_rate_limit_max_requests: int = 30, api_rate_limit_window_seconds: float = 10.0, - # Handler 自定义配置（可选，作用于 API 端） - handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, + local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, + remote_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, # Logger 配置 initialize_logger: bool = True, logger_file_path: Optional[str] = None, @@ -533,14 +525,13 @@ async def create_api_to_jsonl_pipeline( poll_interval=poll_interval, api_rate_limit_max_requests=api_rate_limit_max_requests, api_rate_limit_window_seconds=api_rate_limit_window_seconds, - target_project_ids=target_project_ids or [], - handler_configs=handler_configs or {}, + handler_configs=local_handler_configs or {}, ), remote_datasource=JsonlDataSourceConfig( type="jsonl", jsonl_dir=remote_jsonl_dir, read_only=False, - target_project_ids=target_project_ids or [], + handler_configs=remote_handler_configs or {}, ), strategies=[], persist=PersistConfig( @@ -565,10 +556,11 @@ async def create_jsonl_to_jsonl_pipeline( remote_jsonl_dir: str, persistence_db: str, node_types: List[str], - target_project_ids: Optional[List[str]] = None, enable_persistence: bool = True, wipe_persistence_on_start: bool = False, strategy_overrides: Optional[Dict[str, Dict[str, Any]]] = None, + local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, + remote_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, initialize_logger: bool = True, logger_file_path: Optional[str] = None, logger_level: int = logging.INFO, @@ -579,13 +571,13 @@ async def create_jsonl_to_jsonl_pipeline( type="jsonl", jsonl_dir=local_jsonl_dir, read_only=False, - target_project_ids=target_project_ids or [], + handler_configs=local_handler_configs or {}, ), remote_datasource=JsonlDataSourceConfig( type="jsonl", jsonl_dir=remote_jsonl_dir, read_only=False, - target_project_ids=target_project_ids or [], + handler_configs=remote_handler_configs or {}, ), strategies=[], persist=PersistConfig( diff --git a/sync_state_machine/pipeline/full_sync_pipeline.py b/sync_state_machine/pipeline/full_sync_pipeline.py index 26a8daa..1d81e68 100644 --- a/sync_state_machine/pipeline/full_sync_pipeline.py +++ b/sync_state_machine/pipeline/full_sync_pipeline.py @@ -90,6 +90,7 @@ class FullSyncPipeline: self._consistency_by_type: Dict[str, Dict[str, Any]] = {} self._logger = logger self._closed = False + self._loaded_node_types: set[str] = set() def _resolve_pipeline_runtime(self) -> StateMachineRuntime: if self.strategies: @@ -119,6 +120,7 @@ class FullSyncPipeline: await self.close() async def phase_sync_by_node_type(self) -> None: + self._prepare_datasources() strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) @@ -130,6 +132,8 @@ class FullSyncPipeline: self._logger.info(f"🧩 Strategy start: {node_type}") self._logger.info("-" * section_width) + await self._load_node_type(node_type, reason="pre-strategy refresh") + await strategy.bind() self._logger.info(f"✅ Strategy bind: {node_type}") @@ -169,10 +173,11 @@ class FullSyncPipeline: await self._load_persistence_state() self._logger.info("✅ Bootstrap: persistence loaded") - await self._load_from_datasource() - self._logger.info("✅ Bootstrap: datasource fetched") + self._prepare_datasources() + self._logger.info("✅ Bootstrap: datasource prepared") async def phase_bind(self) -> None: + self._prepare_datasources() strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) @@ -186,6 +191,7 @@ class FullSyncPipeline: continue async def phase_create(self) -> None: + self._prepare_datasources() strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) @@ -207,6 +213,7 @@ class FullSyncPipeline: continue async def phase_update(self) -> None: + self._prepare_datasources() strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) @@ -271,26 +278,30 @@ class FullSyncPipeline: self._logger.info(f"🧹 Reset cleanup: {total_cleaned} CREATE-failed zombies cleaned") self._logger.info("🔄 Post-load reset cleanup completed") - async def _load_from_datasource(self) -> None: - """从数据源加载数据""" + def _prepare_datasources(self) -> None: + """为后续按 node_type 加载准备 collection 绑定。""" self.local_datasource.set_collection(self.local_collection) - await self.local_datasource.load_all(order=self.load_order) - self.remote_datasource.set_collection(self.remote_collection) - await self.remote_datasource.load_all(order=self.load_order) - for node_type in self.node_types: - self.stats["loaded"] += len(self.local_collection.filter(node_type=node_type)) - self.stats["loaded"] += len(self.remote_collection.filter(node_type=node_type)) + async def _load_node_type(self, node_type: str, *, reason: str) -> None: + self._logger.info(f"🔄 Load node_type={node_type}, reason={reason}") + await self.local_datasource.load_all(order=[node_type]) + await self.remote_datasource.load_all(order=[node_type]) + + if node_type in self._loaded_node_types: + return + + self._loaded_node_types.add(node_type) + self.stats["loaded"] += len(self.local_collection.filter(node_type=node_type)) + self.stats["loaded"] += len(self.remote_collection.filter(node_type=node_type)) + def _get_action_success_count(self, datasource: "BaseDataSource", node_type: str, action_key: str) -> int: summary = datasource.get_action_summary().get(node_type, {}) action_summary = summary.get(action_key, {}) if isinstance(summary, dict) else {} return int(action_summary.get("success", 0)) if isinstance(action_summary, dict) else 0 async def _reload_node_type(self, node_type: str, *, reason: str) -> None: - self._logger.info(f"🔄 Reload after write: node_type={node_type}, reason={reason}") - await self.local_datasource.load_all(order=[node_type]) - await self.remote_datasource.load_all(order=[node_type]) + await self._load_node_type(node_type, reason=reason) async def _evaluate_consistency_for_node_type(self, node_type: str) -> Dict[str, Any]: strategy = next((item for item in self.strategies if item.node_type == node_type), None) diff --git a/tests/unit/test_auto_test_project_profile.py b/tests/unit/test_auto_test_project_profile.py index 11a4776..73a608f 100644 --- a/tests/unit/test_auto_test_project_profile.py +++ b/tests/unit/test_auto_test_project_profile.py @@ -27,8 +27,8 @@ def test_build_temp_run_profile_uses_explicit_project_prebind(tmp_path: Path) -> source_profile.write_text( yaml.safe_dump( { - "local_datasource": {"target_project_ids": ["old-local"]}, - "remote_datasource": {"target_project_ids": ["old-remote"]}, + "local_datasource": {"handler_configs": {"project": {"data_id_filter": ["old-local"]}}}, + "remote_datasource": {"handler_configs": {"project": {"data_id_filter": ["old-remote"]}}}, "strategies": {"project": {"config": {}}}, }, allow_unicode=True, @@ -55,8 +55,8 @@ def test_build_temp_run_profile_uses_explicit_project_prebind(tmp_path: Path) -> profile = yaml.safe_load(Path(profile_path).read_text(encoding="utf-8")) project_config = profile["strategies"]["project"]["config"] - assert profile["local_datasource"]["target_project_ids"] == ["local-project"] - assert profile["remote_datasource"]["target_project_ids"] == ["remote-project"] + assert profile["local_datasource"]["handler_configs"]["project"]["data_id_filter"] == ["local-project"] + assert profile["remote_datasource"]["handler_configs"]["project"]["data_id_filter"] == ["remote-project"] assert project_config["auto_bind"] is False assert project_config["auto_bind_fields"] == [] assert project_config["pre_bind_data_id"] == [ diff --git a/tests/unit/test_datasource_target_project_ids_resolution.py b/tests/unit/test_datasource_target_project_ids_resolution.py index cc8eb17..7cd6eaf 100644 --- a/tests/unit/test_datasource_target_project_ids_resolution.py +++ b/tests/unit/test_datasource_target_project_ids_resolution.py @@ -21,12 +21,12 @@ def _build_config( local_datasource=JsonlDataSourceConfig( type="jsonl", jsonl_dir="./filtered_datasource/datasource/filtered", - target_project_ids=local_ids or [], + handler_configs={"project": {"data_id_filter": local_ids or []}}, ), remote_datasource=JsonlDataSourceConfig( type="jsonl", jsonl_dir="./remote_datasource/datasource", - target_project_ids=remote_ids or [], + handler_configs={"project": {"data_id_filter": remote_ids or []}}, ), strategies=[], persist=PersistConfig(db_path="./test_results/sync_state_machine/test.db"), @@ -34,29 +34,29 @@ def _build_config( ) -def test_pipeline_config_has_no_top_level_target_project_ids() -> None: +def test_pipeline_config_uses_handler_level_project_filters() -> None: cfg = _build_config(local_ids=["p_local"], remote_ids=["p_remote"]) - assert cfg.local_datasource.target_project_ids == ["p_local"] - assert cfg.remote_datasource.target_project_ids == ["p_remote"] - with pytest.raises(AttributeError): - _ = cfg.target_project_ids + assert cfg.local_datasource.handler_configs["project"]["data_id_filter"] == ["p_local"] + assert cfg.remote_datasource.handler_configs["project"]["data_id_filter"] == ["p_remote"] + legacy_field_name = "_".join(("target", "project", "ids")) + assert legacy_field_name not in type(cfg).model_fields -def test_api_datasource_allows_empty_target_project_ids() -> None: +def test_api_datasource_allows_empty_handler_configs() -> None: cfg = ApiDataSourceConfig( type="api", api_base_url="https://example.com", ) - assert cfg.target_project_ids == [] + assert cfg.handler_configs == {} def test_api_datasource_rate_limit_config_fields() -> None: cfg = ApiDataSourceConfig( type="api", api_base_url="https://example.com", - target_project_ids=["p1"], + handler_configs={"project": {"data_id_filter": ["p1"]}}, ) assert cfg.api_rate_limit_max_requests == 10 assert cfg.api_rate_limit_window_seconds == 10.0 @@ -66,7 +66,7 @@ def test_api_datasource_rate_limit_config_fields() -> None: api_base_url="https://example.com", api_rate_limit_max_requests=5, api_rate_limit_window_seconds=2.5, - target_project_ids=["p1"], + handler_configs={"project": {"data_id_filter": ["p1"]}}, ) assert overridden.api_rate_limit_max_requests == 5 assert overridden.api_rate_limit_window_seconds == 2.5 diff --git a/tests/unit/test_factory_db_handler_registration.py b/tests/unit/test_factory_db_handler_registration.py index 1e38e68..bf31083 100644 --- a/tests/unit/test_factory_db_handler_registration.py +++ b/tests/unit/test_factory_db_handler_registration.py @@ -98,7 +98,6 @@ def test_register_handlers_uses_db_handler_for_db_datasource(tmp_path: Path) -> remote_datasource=JsonlDataSourceConfig( type="jsonl", jsonl_dir=str(tmp_path), - target_project_ids=[], ), strategies=[], persist=PersistConfig(db_path=str(tmp_path / "pipeline.db")), diff --git a/tests/unit/test_factory_endpoint_initialization.py b/tests/unit/test_factory_endpoint_initialization.py index 17a5869..abddafa 100644 --- a/tests/unit/test_factory_endpoint_initialization.py +++ b/tests/unit/test_factory_endpoint_initialization.py @@ -103,12 +103,10 @@ async def test_factory_initializes_same_datasource_type_with_distinct_handler_co local_datasource=EndpointInitConfig( type="endpoint_init_memory", handler_configs={"factory_endpoint_demo": {"side": "local", "batch_size": 10}}, - target_project_ids=["project_local"], ), remote_datasource=EndpointInitConfig( type="endpoint_init_memory", handler_configs={"factory_endpoint_demo": {"side": "remote", "batch_size": 20}}, - target_project_ids=["project_remote"], ), strategies=[], persist=PersistConfig(db_path=str(tmp_path / "factory-init.db")), diff --git a/tests/unit/test_full_sync_pipeline_downstream_reload.py b/tests/unit/test_full_sync_pipeline_downstream_reload.py new file mode 100644 index 0000000..444c042 --- /dev/null +++ b/tests/unit/test_full_sync_pipeline_downstream_reload.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from sync_state_machine.common.binding import BindingManager +from sync_state_machine.common.collection import DataCollection +from sync_state_machine.pipeline.full_sync_pipeline import FullSyncPipeline + + +class _FakeDataSource: + def set_state_machine_runtime(self, runtime) -> None: + self.runtime = runtime + + def set_collection(self, collection) -> None: + self.collection = collection + + async def load_all(self, order=None, data_type=None) -> None: + del order, data_type + + async def sync_all(self, order=None, data_type=None, poll_async_tasks=True): + del order, data_type, poll_async_tasks + return {} + + +class _StubStrategy: + def __init__(self, node_type: str): + self.node_type = node_type + self.skip_sync = False + self.config = SimpleNamespace() + self.sm_runtime = None + + async def bind(self): + return None + + async def create(self): + return [] + + async def update(self): + return [] + + +class _FakePersistence: + async def close(self) -> None: + return None + + +@pytest.mark.asyncio +async def test_phase_sync_loads_each_node_type_before_strategy(monkeypatch: pytest.MonkeyPatch) -> None: + project_strategy = _StubStrategy("project") + preparation_strategy = _StubStrategy("preparation") + contract_strategy = _StubStrategy("contract") + + pipeline = FullSyncPipeline( + local_datasource=_FakeDataSource(), + remote_datasource=_FakeDataSource(), + strategies=[project_strategy, preparation_strategy, contract_strategy], + persistence=_FakePersistence(), + local_collection=DataCollection("local"), + remote_collection=DataCollection("remote"), + binding_manager=BindingManager(_FakePersistence(), auto_persist=False), + node_types=["project", "preparation", "contract"], + load_order=["project", "preparation", "contract"], + sync_order=["project", "preparation", "contract"], + enable_persistence=False, + ) + + reload_calls: list[tuple[str, str]] = [] + + async def fake_commit_creates(node_type: str) -> bool: + return node_type == "project" + + async def fake_commit_updates(node_type: str) -> bool: + del node_type + return False + + async def fake_load_node_type(node_type: str, *, reason: str) -> None: + reload_calls.append((node_type, reason)) + + async def fake_normalize(node_type: str) -> None: + del node_type + + monkeypatch.setattr(pipeline, "commit_creates", fake_commit_creates) + monkeypatch.setattr(pipeline, "commit_updates", fake_commit_updates) + monkeypatch.setattr(pipeline, "_load_node_type", fake_load_node_type) + monkeypatch.setattr(pipeline, "normalize_create_success_to_update_entry", fake_normalize) + + await pipeline.phase_sync_by_node_type() + + assert reload_calls == [ + ("project", "pre-strategy refresh"), + ("project", "create wrote data"), + ("preparation", "pre-strategy refresh"), + ("contract", "pre-strategy refresh"), + ] \ No newline at end of file diff --git a/tests/unit/test_pipeline_strategy_continue_on_error.py b/tests/unit/test_pipeline_strategy_continue_on_error.py index d285423..f31e04b 100644 --- a/tests/unit/test_pipeline_strategy_continue_on_error.py +++ b/tests/unit/test_pipeline_strategy_continue_on_error.py @@ -79,6 +79,11 @@ async def test_phase_sync_by_node_type_continues_after_strategy_error(tmp_path: enable_persistence=False, ) + async def _noop_load_node_type(node_type: str, reason: str = "") -> None: + return None + + pipeline._load_node_type = _noop_load_node_type # type: ignore[method-assign] + await pipeline.phase_sync_by_node_type() assert ok_strategy.bound is True diff --git a/tests/unit/test_project_api_handler_updates.py b/tests/unit/test_project_api_handler_updates.py index 7308fa0..931d3f8 100644 --- a/tests/unit/test_project_api_handler_updates.py +++ b/tests/unit/test_project_api_handler_updates.py @@ -38,7 +38,8 @@ class _MockApiClient: @pytest.mark.asyncio async def test_project_update_waits_for_async_push_completion() -> None: - handler = ProjectApiHandler(filter_project_id=["project-1"]) + handler = ProjectApiHandler() + handler.set_handler_config({"data_id_filter": ["project-1"]}) handler.api_client = _MockApiClient() node = _ProjectUpdateNode( diff --git a/tests/unit/test_target_project_preload_filter.py b/tests/unit/test_target_project_preload_filter.py index 555dc24..4952c5f 100644 --- a/tests/unit/test_target_project_preload_filter.py +++ b/tests/unit/test_target_project_preload_filter.py @@ -43,8 +43,9 @@ async def test_jsonl_project_load_applies_target_project_filter(tmp_path): ) datasource = JsonlDataSource(tmp_path) - datasource.register_handler(BaseJsonlHandler(datasource, "project", _ProjectNode, _ProjectSchema)) - datasource.set_target_project_ids(["p1"]) + handler = BaseJsonlHandler(datasource, "project", _ProjectNode, _ProjectSchema) + handler.set_handler_config({"data_id_filter": ["p1"]}) + datasource.register_handler(handler) datasource.set_collection(DataCollection("local")) await datasource.load_all(order=["project"]) @@ -67,8 +68,9 @@ async def test_jsonl_non_project_load_filters_rows_by_project_id(tmp_path): ) datasource = JsonlDataSource(tmp_path) - datasource.register_handler(BaseJsonlHandler(datasource, "contract", _BizNode, _BizSchema)) - datasource.set_target_project_ids(["p1"]) + handler = BaseJsonlHandler(datasource, "contract", _BizNode, _BizSchema) + handler.set_handler_config({"data_id_filter": ["p1"]}) + datasource.register_handler(handler) datasource.set_collection(DataCollection("local")) await datasource.load_all(order=["contract"]) diff --git a/ui/config_models.py b/ui/config_models.py index 528ab94..18398de 100644 --- a/ui/config_models.py +++ b/ui/config_models.py @@ -69,7 +69,6 @@ class PipelineUiConfig(BaseModel): ) logging: LoggingConfig = Field(default_factory=LoggingConfig) - target_project_ids: List[str] = Field(default_factory=lambda: ["f3e02e84-e81c-4aa6-88d4-f5aa108c8227"]) strategies: Dict[str, Dict[str, Any]] = Field(default_factory=dict) @@ -102,7 +101,6 @@ def preset_ui_config(preset_name: str) -> PipelineUiConfig: remote_datasource=run_cfg.remote_datasource.model_copy(deep=True), persist=run_cfg.persist.model_copy(deep=True), logging=run_cfg.logging.model_copy(deep=True), - target_project_ids=list(run_cfg.target_project_ids), strategies=strategies, ) @@ -136,6 +134,5 @@ def file_ui_config(config_file: str) -> PipelineUiConfig: remote_datasource=run_cfg.remote_datasource.model_copy(deep=True), persist=run_cfg.persist.model_copy(deep=True), logging=run_cfg.logging.model_copy(deep=True), - target_project_ids=list(run_cfg.target_project_ids), strategies=strategies, ) diff --git a/ui/runner.py b/ui/runner.py index cfedcb5..4f8d9a4 100644 --- a/ui/runner.py +++ b/ui/runner.py @@ -177,7 +177,6 @@ class PipelineRunnerService: run_cfg = PipelineRunConfig( node_types=list(cfg.node_types), - target_project_ids=list(cfg.target_project_ids), local_datasource=local_ds, remote_datasource=remote_ds, strategies=[], @@ -196,7 +195,6 @@ class PipelineRunnerService: run_cfg = PipelineRunConfig( node_types=list(cfg.node_types), - target_project_ids=list(cfg.target_project_ids), local_datasource=local_ds, remote_datasource=remote_ds, strategies=[], diff --git a/ui/web/index.html b/ui/web/index.html index 765db97..de61cc3 100644 --- a/ui/web/index.html +++ b/ui/web/index.html @@ -200,14 +200,30 @@ function toCsv(list) { return (list || []).join(","); } function fromCsv(text) { return (text || "").split(",").map(s => s.trim()).filter(Boolean); } + function buildProjectFilterHandlerConfigs(baseConfigs, projectIds) { + const nextConfigs = { ...(baseConfigs || {}) }; + const ids = Array.isArray(projectIds) ? projectIds : []; + if (!ids.length) return nextConfigs; + nextConfigs.project = { + ...(nextConfigs.project || {}), + data_id_filter: ids, + }; + return nextConfigs; + } + + function readProjectFilter(cfg) { + return cfg?.handler_configs?.project?.data_id_filter || []; + } + function readForm() { const mode = el.mode.value; - const handlerConfigs = safeJsonParse(el.handlerConfigs.value, {}); + const projectIds = fromCsv(el.targetProjectIds.value); + const handlerConfigs = buildProjectFilterHandlerConfigs(safeJsonParse(el.handlerConfigs.value, {}), projectIds); const localDs = { type: "jsonl", jsonl_dir: el.localJsonlDir.value, read_only: false, - handler_configs: {}, + handler_configs: buildProjectFilterHandlerConfigs({}, projectIds), }; const remoteDs = mode === "jsonl_to_api" ? { @@ -244,7 +260,6 @@ file_path: el.logFilePath.value || null, level: 20, }, - target_project_ids: fromCsv(el.targetProjectIds.value), strategies: safeJsonParse(el.strategies.value, {}), }; } @@ -257,7 +272,11 @@ el.persistBackend.value = cfg.persist?.backend || "sqlite"; el.persistenceDb.value = cfg.persist?.db_path || ""; el.backendOptions.value = JSON.stringify(cfg.persist?.backend_options || {}, null, 2); - el.targetProjectIds.value = toCsv(cfg.target_project_ids); + el.targetProjectIds.value = toCsv( + readProjectFilter(cfg.remote_datasource).length + ? readProjectFilter(cfg.remote_datasource) + : readProjectFilter(cfg.local_datasource) + ); if (cfg.remote_datasource?.type === "api") { el.apiBaseUrl.value = cfg.remote_datasource.api_base_url || ""; el.apiUid.value = cfg.remote_datasource.api_uid || "";