From 966f2e12707f0e8cd3546f8f7f49813d783b09ea Mon Sep 17 00:00:00 2001 From: strepsiades Date: Tue, 7 Apr 2026 14:45:19 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E6=AD=A5=E5=AE=9E=E7=8E=B0=E5=A4=9Apr?= =?UTF-8?q?oject=20pipeline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sync_state_machine/common/binding.py | 73 +++- sync_state_machine/common/collection.py | 119 +++++- .../common/persistence_service.py | 85 +++++ .../config/multi_project_config.py | 1 + sync_state_machine/logging.py | 12 + sync_state_machine/pipeline/factory.py | 169 +++++++-- .../pipeline/full_sync_pipeline.py | 70 ++-- .../pipeline/multi_project_pipeline.py | 349 ++++++++++++++---- .../pipeline/project_scope_runtime.py | 102 +++++ .../test_multi_project_pipeline_config.py | 249 ++++++++++++- tests/unit/test_summary_report.py | 23 +- 11 files changed, 1083 insertions(+), 169 deletions(-) create mode 100644 sync_state_machine/common/persistence_service.py create mode 100644 sync_state_machine/pipeline/project_scope_runtime.py diff --git a/sync_state_machine/common/binding.py b/sync_state_machine/common/binding.py index 943948f..ec633b9 100644 --- a/sync_state_machine/common/binding.py +++ b/sync_state_machine/common/binding.py @@ -4,6 +4,7 @@ from typing import Optional, Dict, Any, List from ..logging import get_logger from .persistence import PersistenceBackend +from .persistence_service import ScopedPersistenceView from .types import BindingStatus, BindingRecord logger = get_logger(__name__) @@ -14,14 +15,53 @@ class BindingManager: BindingManager 负责维护不同 DataCollection 间节点的绑定关系。 所有查询均在内存中进行，持久化仅作为备份。 """ - def __init__(self, persistence: PersistenceBackend, auto_persist: bool = False, scope: str = "global"): + def __init__( + self, + persistence: PersistenceBackend | ScopedPersistenceView, + auto_persist: bool = False, + scope: str = "global", + global_fallback_manager: Optional["BindingManager"] = None, + ): self.persistence = persistence self.auto_persist = auto_persist self.scope = scope + self._global_fallback_manager = global_fallback_manager # { node_type: { local_id: remote_id } } self._bindings: Dict[str, Dict[str, Optional[str]]] = {} self._deleted: Dict[str, set[str]] = {} + def set_global_fallback_manager(self, manager: Optional["BindingManager"]) -> None: + self._global_fallback_manager = manager + + async def _persistence_save_binding(self, node_type: str, local_id: str, payload_dict: Dict[str, Any]) -> None: + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.save_binding(node_type, local_id, payload_dict) + return + await self.persistence.save_binding(self.scope, node_type, local_id, payload_dict) + + async def _persistence_delete_binding(self, node_type: str, local_id: str) -> None: + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.delete_binding(node_type, local_id) + return + await self.persistence.delete_binding(self.scope, node_type, local_id) + + async def _persistence_delete_bindings_bulk(self, node_type: str, local_ids: List[str]) -> None: + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.delete_bindings_bulk(node_type, local_ids) + return + await self.persistence.delete_bindings_bulk(self.scope, node_type, local_ids) + + async def _persistence_save_bindings_bulk(self, node_type: str, bindings: Dict[str, Optional[str]]) -> None: + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.save_bindings_bulk(node_type, bindings) + return + await self.persistence.save_bindings_bulk(self.scope, node_type, bindings) + + async def _persistence_load_bindings(self, node_type: str) -> List[Dict[str, Any]]: + if isinstance(self.persistence, ScopedPersistenceView): + return await self.persistence.load_bindings(node_type) + return await self.persistence.load_bindings(self.scope, node_type) + def _get_type_bindings(self, node_type: str) -> Dict[str, Optional[str]]: if node_type not in self._bindings: self._bindings[node_type] = {} @@ -58,7 +98,7 @@ class BindingManager: logger.debug(log_line) if self.auto_persist: - await self.persistence.save_binding(self.scope, node_type, local_id, {"remote_id": remote_id}) + await self._persistence_save_binding(node_type, local_id, {"remote_id": remote_id}) async def unbind(self, node_type: str, local_id: str, manual: bool = False): """ @@ -83,14 +123,19 @@ class BindingManager: ) if self.auto_persist: - await self.persistence.delete_binding(self.scope, node_type, local_id) + await self._persistence_delete_binding(node_type, local_id) else: self._get_deleted(node_type).add(local_id) async def get_remote_id(self, node_type: str, local_id: str) -> Optional[str]: """查询本地节点对应的远程 ID""" bindings = self._get_type_bindings(node_type) - return bindings.get(local_id) + remote_id = bindings.get(local_id) + if remote_id is not None or local_id in bindings: + return remote_id + if self._global_fallback_manager is None: + return None + return await self._global_fallback_manager.get_remote_id(node_type, local_id) async def get_local_id(self, node_type: str, remote_id: str) -> Optional[str]: """查询远程节点对应的本地 ID (内存遍历)""" @@ -98,11 +143,13 @@ class BindingManager: for lid, rid in bindings.items(): if rid == remote_id: return lid - return None + if self._global_fallback_manager is None: + return None + return await self._global_fallback_manager.get_local_id(node_type, remote_id) async def get_record(self, node_type: str, local_id: str) -> Optional[BindingRecord]: """获取结构化绑定记录""" - remote_id = self._get_type_bindings(node_type).get(local_id) + remote_id = await self.get_remote_id(node_type, local_id) if remote_id is None: return None return BindingRecord( @@ -122,7 +169,7 @@ class BindingManager: async def load_from_persistence(self, node_type: str): """从持久化层加载特定类型的绑定关系到内存""" - raw_bindings = await self.persistence.load_bindings(self.scope, node_type) + raw_bindings = await self._persistence_load_bindings(node_type) bindings = self._get_type_bindings(node_type) bindings.clear() self._get_deleted(node_type).clear() @@ -165,7 +212,7 @@ class BindingManager: del bindings[local_id] deleted_count += 1 if self.auto_persist: - await self.persistence.delete_binding(self.scope, node_type, local_id) + await self._persistence_delete_binding(node_type, local_id) else: self._get_deleted(node_type).add(local_id) @@ -198,13 +245,19 @@ class BindingManager: touched_types.append(node_type) if deleted: - await self.persistence.delete_bindings_bulk(self.scope, node_type, deleted) + await self._persistence_delete_bindings_bulk(node_type, deleted) self._get_deleted(node_type).clear() if bindings: - await self.persistence.save_bindings_bulk(self.scope, node_type, bindings) + await self._persistence_save_bindings_bulk(node_type, bindings) touched_text = ",".join(touched_types) if touched_types else "none" logger.info( f"🔍 [BindingManager.persist] 完成: scope={self.scope}, types={type_count}, saved={saved_total}, deleted={deleted_total}, touched=[{touched_text}]" ) + + async def unload(self, *, node_types: Optional[List[str]] = None) -> None: + target_types = list(node_types) if node_types is not None else list(self._bindings.keys()) + for node_type in target_types: + self._bindings.pop(node_type, None) + self._deleted.pop(node_type, None) diff --git a/sync_state_machine/common/collection.py b/sync_state_machine/common/collection.py index 70b1d8e..42375dc 100644 --- a/sync_state_machine/common/collection.py +++ b/sync_state_machine/common/collection.py @@ -5,6 +5,7 @@ from typing import Dict, List, Optional, Any, Set, Callable, Literal from .sync_node import SyncNode from .types import SyncStatus from .persistence import PersistenceBackend +from .persistence_service import ScopedPersistenceView class DataCollection: @@ -24,15 +25,17 @@ class DataCollection: self, collection_id: str, scope: str = "global", - persistence: Optional[PersistenceBackend] = None, + persistence: Optional[PersistenceBackend | ScopedPersistenceView] = None, auto_persist: bool = False, sm_runtime: Optional[Any] = None, + global_fallback_collection: Optional["DataCollection"] = None, ): self.collection_id = collection_id self.scope = scope self.persistence = persistence self.auto_persist = auto_persist self._sm_runtime = sm_runtime + self._global_fallback_collection = global_fallback_collection self._nodes: Dict[str, SyncNode] = {} # 内存缓存 # data_id 全局唯一（按 collection 维度），用于 O(1) 反查 node_id self._data_id_to_node_id: Dict[str, str] = {} @@ -53,6 +56,69 @@ class DataCollection: """设置 Collection 默认状态机 runtime。""" self._sm_runtime = runtime + def set_global_fallback_collection(self, collection: Optional["DataCollection"]) -> None: + self._global_fallback_collection = collection + + def _get_global_fallback(self) -> Optional["DataCollection"]: + return self._global_fallback_collection + + async def _persistence_save_node(self, node: SyncNode) -> None: + if not self.persistence: + return + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.save_node(self.collection_id, node) + return + await self.persistence.save_node(self.collection_id, self.scope, node) + + async def _persistence_delete_node(self, node_id: str) -> None: + if not self.persistence: + return + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.delete_node(self.collection_id, node_id) + return + await self.persistence.delete_node(self.collection_id, self.scope, node_id) + + async def _persistence_delete_nodes_bulk(self, node_ids: List[str]) -> None: + if not self.persistence: + return + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.delete_nodes_bulk(self.collection_id, node_ids) + return + await self.persistence.delete_nodes_bulk(self.collection_id, self.scope, node_ids) + + async def _persistence_save_nodes_bulk(self, nodes: List[SyncNode]) -> None: + if not self.persistence: + return + if isinstance(self.persistence, ScopedPersistenceView): + await self.persistence.save_nodes_bulk(self.collection_id, nodes) + return + await self.persistence.save_nodes_bulk(self.collection_id, self.scope, nodes) + + async def _persistence_load_nodes( + self, + *, + exclude_node_types: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: + if not self.persistence: + return [] + if isinstance(self.persistence, ScopedPersistenceView): + return await self.persistence.load_nodes( + self.collection_id, + exclude_node_types=exclude_node_types, + ) + return await self.persistence.load_nodes( + self.collection_id, + self.scope, + exclude_node_types=exclude_node_types, + ) + + async def _persistence_load_bind_data_id_overrides(self, node_type: str) -> Dict[str, Optional[str]]: + if not self.persistence: + return {} + if isinstance(self.persistence, ScopedPersistenceView): + return await self.persistence.load_bind_data_id_overrides(node_type) + return await self.persistence.load_bind_data_id_overrides(node_type, self.scope) + def _remove_data_id_index(self, node_id: str, data_id: Optional[str] = None) -> None: target_data_id = data_id if target_data_id is None: @@ -86,7 +152,7 @@ class DataCollection: self._nodes[node.node_id] = node self._deleted_node_ids.discard(node.node_id) if self.persistence and self.auto_persist: - await self.persistence.save_node(self.collection_id, self.scope, node) + await self._persistence_save_node(node) async def update(self, node: SyncNode): old_data_id = self._node_id_to_data_id.get(node.node_id) @@ -105,7 +171,7 @@ class DataCollection: self._nodes[node.node_id] = node self._deleted_node_ids.discard(node.node_id) if self.persistence and self.auto_persist: - await self.persistence.save_node(self.collection_id, self.scope, node) + await self._persistence_save_node(node) async def delete(self, node_id: str): self._remove_data_id_index(node_id) @@ -114,12 +180,18 @@ class DataCollection: del self._nodes[node_id] if self.persistence: if self.auto_persist: - await self.persistence.delete_node(self.collection_id, self.scope, node_id) + await self._persistence_delete_node(node_id) else: self._deleted_node_ids.add(node_id) def get(self, node_id: str) -> Optional[SyncNode]: - return self._nodes.get(node_id) + node = self._nodes.get(node_id) + if node is not None: + return node + fallback = self._get_global_fallback() + if fallback is None: + return None + return fallback.get(node_id) def get_by_node_id(self, node_id: str) -> Optional[SyncNode]: """别名方法，与 get() 功能相同""" @@ -131,14 +203,23 @@ class DataCollection: def get_node_id_by_data_id(self, data_id: str) -> Optional[str]: """按 data_id 反查 node_id（O(1)）。""" - return self._data_id_to_node_id.get(data_id) + node_id = self._data_id_to_node_id.get(data_id) + if node_id is not None: + return node_id + fallback = self._get_global_fallback() + if fallback is None: + return None + return fallback.get_node_id_by_data_id(data_id) def get_by_data_id_any(self, data_id: str) -> Optional[SyncNode]: """按 data_id 查找节点（不要求 node_type）（O(1)）。""" node_id = self._data_id_to_node_id.get(data_id) - if not node_id: + if node_id: + return self._nodes.get(node_id) + fallback = self._get_global_fallback() + if fallback is None: return None - return self._nodes.get(node_id) + return fallback.get_by_data_id_any(data_id) def get_by_data_id(self, node_type: str, data_id: str) -> Optional[SyncNode]: """按 node_type 和 data_id 查找节点""" @@ -389,7 +470,7 @@ class DataCollection: self._node_id_to_data_id = {} self._deleted_node_ids.clear() - node_dicts = await self.persistence.load_nodes(self.collection_id, self.scope) + node_dicts = await self._persistence_load_nodes() from .types import SyncAction, SyncStatus, BindingStatus from .registry import DomainRegistry @@ -511,7 +592,7 @@ class DataCollection: node_types = {node.node_type for node in self._nodes.values()} for node_type in node_types: - override_map = await self.persistence.load_bind_data_id_overrides(node_type, self.scope) + override_map = await self._persistence_load_bind_data_id_overrides(node_type) for node in self.filter(node_type=node_type): bind_data_id = override_map.get(node.node_id) if bind_data_id: @@ -529,11 +610,7 @@ class DataCollection: self._node_id_to_data_id = {} self._deleted_node_ids.clear() - node_dicts = await self.persistence.load_nodes( - self.collection_id, - self.scope, - exclude_node_types=node_types, - ) + node_dicts = await self._persistence_load_nodes(exclude_node_types=node_types) from .types import SyncAction, SyncStatus, BindingStatus from .registry import DomainRegistry @@ -637,7 +714,7 @@ class DataCollection: loaded_node_types = {node.node_type for node in self._nodes.values()} for node_type in loaded_node_types: - override_map = await self.persistence.load_bind_data_id_overrides(node_type, self.scope) + override_map = await self._persistence_load_bind_data_id_overrides(node_type) for node in self.filter(node_type=node_type): bind_data_id = override_map.get(node.node_id) if bind_data_id: @@ -670,7 +747,7 @@ class DataCollection: # 先落库删除（用于 auto_persist=False 的场景） if self._deleted_node_ids: - await self.persistence.delete_nodes_bulk(self.collection_id, self.scope, list(self._deleted_node_ids)) + await self._persistence_delete_nodes_bulk(list(self._deleted_node_ids)) self._deleted_node_ids.clear() if not self._nodes: @@ -683,7 +760,13 @@ class DataCollection: if not nodes_to_persist: return - await self.persistence.save_nodes_bulk(self.collection_id, self.scope, nodes_to_persist) + await self._persistence_save_nodes_bulk(nodes_to_persist) + + async def unload(self) -> None: + self._nodes.clear() + self._data_id_to_node_id.clear() + self._node_id_to_data_id.clear() + self._deleted_node_ids.clear() async def filter_by_project_ids(self, data_id_filter: List[str]) -> int: """按目标项目 data_id 集合过滤当前 collection，返回删除数量。""" diff --git a/sync_state_machine/common/persistence_service.py b/sync_state_machine/common/persistence_service.py new file mode 100644 index 0000000..13d5f6f --- /dev/null +++ b/sync_state_machine/common/persistence_service.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .persistence import PersistenceBackend + + +class PersistenceService: + """Shared persistence service that owns the backend lifecycle.""" + + def __init__(self, backend: PersistenceBackend): + self._backend = backend + + @property + def backend(self) -> PersistenceBackend: + return self._backend + + async def initialize(self) -> None: + await self._backend.initialize() + + def view(self, scope: str) -> "ScopedPersistenceView": + return ScopedPersistenceView(self._backend, scope=scope) + + async def close(self) -> None: + await self._backend.close() + + +class ScopedPersistenceView: + """Thin scope-bound persistence view used by project runtimes.""" + + def __init__(self, backend: PersistenceBackend, *, scope: str): + self._backend = backend + self.scope = scope + + @property + def backend(self) -> PersistenceBackend: + return self._backend + + async def load_nodes( + self, + collection_id: str, + *, + exclude_node_types: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: + return await self._backend.load_nodes( + collection_id, + self.scope, + exclude_node_types=exclude_node_types, + ) + + async def save_node(self, collection_id: str, node) -> None: + await self._backend.save_node(collection_id, self.scope, node) + + async def save_nodes_bulk(self, collection_id: str, nodes) -> None: + await self._backend.save_nodes_bulk(collection_id, self.scope, nodes) + + async def delete_node(self, collection_id: str, node_id: str) -> None: + await self._backend.delete_node(collection_id, self.scope, node_id) + + async def delete_nodes_bulk(self, collection_id: str, node_ids: List[str]) -> None: + await self._backend.delete_nodes_bulk(collection_id, self.scope, node_ids) + + async def load_bindings(self, node_type: str) -> List[Dict[str, Any]]: + return await self._backend.load_bindings(self.scope, node_type) + + async def save_binding(self, node_type: str, local_id: str, payload_dict: Dict[str, Any]) -> None: + await self._backend.save_binding(self.scope, node_type, local_id, payload_dict) + + async def save_bindings_bulk(self, node_type: str, bindings: Dict[str, Optional[str]]) -> None: + await self._backend.save_bindings_bulk(self.scope, node_type, bindings) + + async def delete_binding(self, node_type: str, local_id: str) -> None: + await self._backend.delete_binding(self.scope, node_type, local_id) + + async def delete_bindings_bulk(self, node_type: str, local_ids: List[str]) -> None: + await self._backend.delete_bindings_bulk(self.scope, node_type, local_ids) + + async def load_bind_data_id_overrides(self, node_type: str) -> Dict[str, Optional[str]]: + return await self._backend.load_bind_data_id_overrides(node_type, self.scope) + + async def delete_scope_node_types(self, *, node_types: List[str]) -> None: + await self._backend.delete_scope_node_types(scope=self.scope, node_types=node_types) + + async def close(self) -> None: + return None \ No newline at end of file diff --git a/sync_state_machine/config/multi_project_config.py b/sync_state_machine/config/multi_project_config.py index 7584e62..99e2bbc 100644 --- a/sync_state_machine/config/multi_project_config.py +++ b/sync_state_machine/config/multi_project_config.py @@ -25,6 +25,7 @@ class MultiProjectRunConfig(BaseModel): global_config: str default_project_config: str + max_concurrency: int = 20 project_bootstrap_node_types: List[str] = [] projects: List[MultiProjectItemConfig] diff --git a/sync_state_machine/logging.py b/sync_state_machine/logging.py index ad77e32..924452a 100644 --- a/sync_state_machine/logging.py +++ b/sync_state_machine/logging.py @@ -8,6 +8,14 @@ from typing import Optional APP_LOGGER_NAME = "sync_state_machine" +class ScopeLoggerAdapter(logging.LoggerAdapter): + def process(self, msg, kwargs): + scope = self.extra.get("scope") + if scope and not str(msg).startswith("["): + msg = f"[{scope}] {msg}" + return msg, kwargs + + class _NodeLogFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: message = record.getMessage() @@ -30,6 +38,10 @@ def get_logger(name: Optional[str] = None) -> logging.Logger: return logging.getLogger(f"{APP_LOGGER_NAME}.{name}") +def get_scope_logger(scope: str, name: Optional[str] = None) -> logging.LoggerAdapter: + return ScopeLoggerAdapter(get_logger(name), {"scope": scope}) + + def clear_app_logger_handlers(logger: Optional[logging.Logger] = None) -> None: target = logger or get_app_logger() handlers = list(target.handlers) diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index 43b6ed2..49acf43 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -17,11 +17,12 @@ import inspect import json import logging from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Literal, Optional from ..common.binding import BindingManager from ..common.collection import DataCollection from ..common.persistence import PersistenceBackend, create_persistence_backend +from ..common.persistence_service import PersistenceService from ..common.registry import DomainRegistry from ..datasource import ensure_builtin_datasource_types_registered from ..datasource.type_registry import get_datasource_factory @@ -47,10 +48,13 @@ from ..config.strategy_config import resolve_domain_option_config from ..sync_system.strategy import BaseSyncStrategy from ..logging import configure_app_logging, ensure_app_logging, get_logger from .full_sync_pipeline import FullSyncPipeline +from .project_scope_runtime import ProjectScopeRuntime logger = get_logger(__name__) +PipelineType = Literal["auto", "full", "multi-project"] + def _project_root() -> Path: return Path(__file__).resolve().parents[2] @@ -229,12 +233,32 @@ def _add_handlers( def _build_collections_and_bindings( *, - persistence: PersistenceBackend, + persistence, scope: str, + local_fallback_collection: Optional[DataCollection] = None, + remote_fallback_collection: Optional[DataCollection] = None, + binding_fallback_manager: Optional[BindingManager] = None, ) -> tuple[DataCollection, DataCollection, BindingManager]: - local_collection = DataCollection("local", scope=scope, persistence=persistence, auto_persist=False) - remote_collection = DataCollection("remote", scope=scope, persistence=persistence, auto_persist=False) - binding_manager = BindingManager(persistence, auto_persist=False, scope=scope) + local_collection = DataCollection( + "local", + scope=scope, + persistence=persistence, + auto_persist=False, + global_fallback_collection=local_fallback_collection, + ) + remote_collection = DataCollection( + "remote", + scope=scope, + persistence=persistence, + auto_persist=False, + global_fallback_collection=remote_fallback_collection, + ) + binding_manager = BindingManager( + persistence, + auto_persist=False, + scope=scope, + global_fallback_manager=binding_fallback_manager, + ) return local_collection, remote_collection, binding_manager @@ -331,22 +355,86 @@ async def create_pipeline_from_config( ephemeral_node_types: Optional[List[str]] = None, local_datasource_override=None, remote_datasource_override=None, + local_fallback_collection: Optional[DataCollection] = None, + remote_fallback_collection: Optional[DataCollection] = None, + binding_fallback_manager: Optional[BindingManager] = None, + persistence_service: Optional[PersistenceService] = None, + scope_runtime: Optional[ProjectScopeRuntime] = None, ) -> FullSyncPipeline: ensure_app_logging(config.logging) all_types = list(config.node_types) strategy_map = _resolve_strategy_map(config, all_types) - db_path = Path(config.persist.db_path) - if config.persist.wipe_on_start and db_path.exists(): - db_path.unlink() + runtime = scope_runtime + close_resources = runtime is None - persistence: Optional[PersistenceBackend] = None + if runtime is None: + runtime = await create_scope_runtime_from_config( + config, + scope=scope, + local_datasource_override=local_datasource_override, + remote_datasource_override=remote_datasource_override, + local_fallback_collection=local_fallback_collection, + remote_fallback_collection=remote_fallback_collection, + binding_fallback_manager=binding_fallback_manager, + persistence_service=persistence_service, + ) + + strategies = _build_strategies( + config, + all_types, + runtime.local_collection, + runtime.remote_collection, + runtime.binding_manager, + strategy_map, + ) + + return FullSyncPipeline( + local_datasource=runtime.local_datasource, + remote_datasource=runtime.remote_datasource, + strategies=strategies, + node_types=all_types, + persistence=runtime.persistence_backend, + local_collection=runtime.local_collection, + remote_collection=runtime.remote_collection, + binding_manager=runtime.binding_manager, + scope=scope, + pipeline_name=pipeline_name, + bootstrap_binding_node_types=bootstrap_binding_node_types, + ephemeral_node_types=ephemeral_node_types, + enable_persistence=config.persist.enable, + scope_runtime=runtime, + close_resources=close_resources, + ) + + +async def create_scope_runtime_from_config( + config: PipelineRunConfig, + *, + scope: str = "global", + local_datasource_override=None, + remote_datasource_override=None, + local_fallback_collection: Optional[DataCollection] = None, + remote_fallback_collection: Optional[DataCollection] = None, + binding_fallback_manager: Optional[BindingManager] = None, + persistence_service: Optional[PersistenceService] = None, + global_runtime: Optional[ProjectScopeRuntime] = None, +) -> ProjectScopeRuntime: + ensure_app_logging(config.logging) + + db_path = Path(config.persist.db_path) + owns_persistence_service = persistence_service is None + created_backend: Optional[PersistenceBackend] = None local_ds = None remote_ds = None try: - persistence = create_persistence_backend(config.persist) - await persistence.initialize() + if persistence_service is None: + if config.persist.wipe_on_start and db_path.exists(): + db_path.unlink() + created_backend = create_persistence_backend(config.persist) + persistence_service = PersistenceService(created_backend) + await persistence_service.initialize() local_ds = await _create_and_initialize_datasource( config.local_datasource, @@ -360,43 +448,34 @@ async def create_pipeline_from_config( endpoint_name="remote", project_root=_project_root(), ) - _add_handlers(config, local_ds, remote_ds, all_types) + _add_handlers(config, local_ds, remote_ds, list(config.node_types)) + persistence_view = persistence_service.view(scope) local_collection, remote_collection, binding_manager = _build_collections_and_bindings( - persistence=persistence, + persistence=persistence_view, scope=scope, + local_fallback_collection=local_fallback_collection, + remote_fallback_collection=remote_fallback_collection, + binding_fallback_manager=binding_fallback_manager, ) - strategies = _build_strategies( - config, - all_types, - local_collection, - remote_collection, - binding_manager, - strategy_map, - ) - - pipeline = FullSyncPipeline( - local_datasource=local_ds, - remote_datasource=remote_ds, - strategies=strategies, - node_types=all_types, - persistence=persistence, + return ProjectScopeRuntime( + scope=scope, + persistence_service=persistence_service, + persistence_view=persistence_view, local_collection=local_collection, remote_collection=remote_collection, binding_manager=binding_manager, - scope=scope, - pipeline_name=pipeline_name, - bootstrap_binding_node_types=bootstrap_binding_node_types, - ephemeral_node_types=ephemeral_node_types, - enable_persistence=config.persist.enable, + local_datasource=local_ds, + remote_datasource=remote_ds, + global_runtime=global_runtime, + owns_persistence_service=owns_persistence_service, ) - - return pipeline except Exception: await _safe_close_resource("remote_datasource", remote_ds) await _safe_close_resource("local_datasource", local_ds) - await _safe_close_resource("persistence", persistence) + if owns_persistence_service: + await _safe_close_resource("persistence", persistence_service or created_backend) raise @@ -409,9 +488,27 @@ async def run_pipeline_from_config( bootstrap_binding_node_types: Optional[List[str]] = None, local_datasource_override=None, remote_datasource_override=None, + pipeline_type: PipelineType = "auto", ) -> Dict[str, Any]: ensure_app_logging(config.logging) + from .multi_project_pipeline import is_implicit_project_batch_config, run_implicit_project_batch_from_config + + implicit_multi_project = is_implicit_project_batch_config(config) + if pipeline_type == "multi-project" and not implicit_multi_project: + raise ValueError("pipeline_type=multi-project requires an implicit multi-project pipeline config") + if pipeline_type == "full" and implicit_multi_project: + logger.info("pipeline_type=full specified, skipping implicit multi-project routing") + + if pipeline_type != "full" and implicit_multi_project: + _log_resolved_config(config) + return await run_implicit_project_batch_from_config( + config, + title=title, + print_summary=print_summary, + max_concurrency=getattr(config, "max_concurrency", None), + ) + _log_resolved_config(config) pipeline: Optional[FullSyncPipeline] = None diff --git a/sync_state_machine/pipeline/full_sync_pipeline.py b/sync_state_machine/pipeline/full_sync_pipeline.py index 640166b..6ad3d80 100644 --- a/sync_state_machine/pipeline/full_sync_pipeline.py +++ b/sync_state_machine/pipeline/full_sync_pipeline.py @@ -15,7 +15,7 @@ if TYPE_CHECKING: from ..datasource.jsonl import JsonlDataSource from ..common.collection import DataCollection from ..common.binding import BindingManager -from ..logging import get_logger +from ..logging import get_scope_logger from ..common.persistence import PersistenceBackend from ..sync_system.config import UpdateDirection from ..sync_system.strategy import BaseSyncStrategy @@ -25,9 +25,7 @@ from ..sync_system.strategy_ops import finalize_peer_creating_sources, run_phase from ..sync_system.strategy_ops.bind_ops import refresh_bind_data_id from ..sync_system.strategy_ops.post_check_ops import evaluate_consistency_for_node_type from .summary_report import print_pipeline_summary - - -logger = get_logger(__name__) +from .project_scope_runtime import ProjectScopeRuntime class NodeTypeLoadError(RuntimeError): @@ -72,6 +70,8 @@ class FullSyncPipeline: load_order: Optional[List[str]] = None, sync_order: Optional[List[str]] = None, enable_persistence: bool = True, + scope_runtime: Optional[ProjectScopeRuntime] = None, + close_resources: bool = True, ): # 验证必需参数（由工厂函数创建） if strategies is None: @@ -91,6 +91,8 @@ class FullSyncPipeline: self.sync_order = sync_order or self.node_types self.persistence = persistence self.enable_persistence = enable_persistence + self.scope_runtime = scope_runtime + self.close_resources = close_resources self.local_collection = local_collection self.remote_collection = remote_collection @@ -113,12 +115,9 @@ class FullSyncPipeline: "post_check_passed": True, } self._consistency_by_type: Dict[str, Dict[str, Any]] = {} - self._logger = logger + self._logger = get_scope_logger(scope, __name__) self._closed = False self._loaded_node_types: set[str] = set() - self._preloaded_local_nodes: List[Any] = [] - self._preloaded_remote_nodes: List[Any] = [] - self._preloaded_bindings: Dict[str, Dict[str, Optional[str]]] = {} def _resolve_pipeline_runtime(self) -> StateMachineRuntime: if self.strategies: @@ -290,12 +289,17 @@ class FullSyncPipeline: continue async def phase_bootstrap(self) -> None: - await self._load_persistence_state() + if self.scope_runtime is not None: + await self.scope_runtime.load( + node_types=self.node_types, + bootstrap_binding_node_types=self.bootstrap_binding_node_types, + enable_persistence=self.enable_persistence, + ephemeral_node_types=self.ephemeral_node_types, + ) + else: + await self._load_persistence_state() self._logger.info("✅ Bootstrap: persistence loaded") - await self._apply_preloaded_state() - self._logger.info("✅ Bootstrap: shared preload applied") - self._prepare_datasources() self._logger.info("✅ Bootstrap: datasource prepared") @@ -383,6 +387,14 @@ class FullSyncPipeline: return # 已经关闭过，避免重复关闭 self._closed = True + if not self.close_resources: + return + + if self.scope_runtime is not None: + await self.scope_runtime.unload() + await self.scope_runtime.close() + return + await self._safe_close("remote_datasource", self.remote_datasource.close()) await self._safe_close("local_datasource", self.local_datasource.close()) await self._safe_close("persistence", self.persistence.close()) @@ -459,28 +471,6 @@ class FullSyncPipeline: if total_cleaned > 0: self._logger.info(f"🧹 Reset cleanup: {total_cleaned} bootstrap zombies cleaned") - async def _apply_preloaded_state(self) -> None: - if self._preloaded_local_nodes: - await self.local_collection.import_nodes(self._preloaded_local_nodes) - if self._preloaded_remote_nodes: - await self.remote_collection.import_nodes(self._preloaded_remote_nodes) - for node_type, bindings in self._preloaded_bindings.items(): - self.binding_manager.import_bindings(node_type, bindings) - - def preload_shared_state( - self, - *, - local_nodes: List[Any], - remote_nodes: List[Any], - bindings_by_type: Dict[str, Dict[str, Optional[str]]], - ) -> None: - self._preloaded_local_nodes = list(local_nodes) - self._preloaded_remote_nodes = list(remote_nodes) - self._preloaded_bindings = { - node_type: dict(bindings) - for node_type, bindings in bindings_by_type.items() - } - def _prepare_datasources(self) -> None: """为后续按 node_type 加载准备 collection 绑定。""" self.local_datasource.set_collection(self.local_collection) @@ -772,6 +762,18 @@ class FullSyncPipeline: async def phase_persistence(self) -> None: """持久化所有数据。""" + if self.scope_runtime is not None: + if not self.enable_persistence: + self._logger.info("⏭️ Persistence disabled, skipping...") + return + self._logger.info("🔍 Persisting scope runtime...") + await self.scope_runtime.persist( + enable_persistence=self.enable_persistence, + ephemeral_node_types=self.ephemeral_node_types, + ) + self._logger.info("✅ All data persisted successfully") + return + if not self.enable_persistence: self._logger.info("⏭️ Persistence disabled, skipping...") return diff --git a/sync_state_machine/pipeline/multi_project_pipeline.py b/sync_state_machine/pipeline/multi_project_pipeline.py index d5cf9d7..c4269d7 100644 --- a/sync_state_machine/pipeline/multi_project_pipeline.py +++ b/sync_state_machine/pipeline/multi_project_pipeline.py @@ -1,28 +1,34 @@ from __future__ import annotations +import asyncio import logging -import copy +from dataclasses import dataclass from pathlib import Path -from typing import Dict, Optional, List, TypedDict, Any +from typing import Any, Dict, Optional, List from ..config import ( MultiProjectItemConfig, MultiProjectRunConfig, PipelineRunConfig, + StrategyRuntimeConfig, apply_config_overrides, build_config_from_file, ) -from ..logging import get_logger -from .factory import create_pipeline_from_config +from ..logging import get_scope_logger +from .factory import create_pipeline_from_config, create_scope_runtime_from_config -logger = get_logger(__name__) +logger = get_scope_logger("multi-project", __name__) -class _SharedStateSnapshot(TypedDict): - local_nodes: List[Any] - remote_nodes: List[Any] - bindings_by_type: Dict[str, Dict[str, Optional[str]]] +@dataclass(frozen=True) +class ImplicitProjectItem: + local_project_id: str + remote_project_id: str + + @property + def scope(self) -> str: + return f"project_id:{self.local_project_id}" def _resolve_embedded_config_path(raw_path: str, *, project_root: Path, profile_path: Path) -> Path: @@ -65,7 +71,215 @@ def _resolve_project_bootstrap_node_types( return list(dict.fromkeys(shared_node_types)) -class MultiProjectPipeline: +def _normalize_project_id_items(raw_items: Any) -> list[str]: + items = raw_items if isinstance(raw_items, list) else [] + normalized: list[str] = [] + for item in items: + value = str(item).strip() + if value and value not in normalized: + normalized.append(value) + return normalized + + +def _get_project_handler_filters(config: PipelineRunConfig) -> tuple[list[str], list[str]]: + local_filters = _normalize_project_id_items( + config.local_datasource.handler_configs.get("project", {}).get("data_id_filter", []) + ) + remote_filters = _normalize_project_id_items( + config.remote_datasource.handler_configs.get("project", {}).get("data_id_filter", []) + ) + return local_filters, remote_filters + + +def _get_project_strategy_runtime(config: PipelineRunConfig) -> StrategyRuntimeConfig | None: + for item in config.strategies: + if item.node_type == "project": + return item + return None + + +def _get_project_prebind_pairs(config: PipelineRunConfig) -> list[tuple[str, str]]: + runtime_cfg = _get_project_strategy_runtime(config) + if runtime_cfg is None: + return [] + pairs: list[tuple[str, str]] = [] + for item in runtime_cfg.config.pre_bind_data_id: + local_id = str(item.local_data_id).strip() + remote_id = str(item.remote_data_id).strip() + if not local_id or not remote_id: + continue + pairs.append((local_id, remote_id)) + return pairs + + +def is_implicit_project_batch_config(config: PipelineRunConfig) -> bool: + if "project" not in config.node_types: + return False + local_filters, remote_filters = _get_project_handler_filters(config) + prebind_pairs = _get_project_prebind_pairs(config) + return max(len(local_filters), len(remote_filters), len(prebind_pairs)) > 1 + + +def _extract_implicit_project_items(config: PipelineRunConfig) -> list[ImplicitProjectItem]: + local_filters, remote_filters = _get_project_handler_filters(config) + prebind_pairs = _get_project_prebind_pairs(config) + if not prebind_pairs: + raise ValueError("implicit multi-project config requires strategies.project.config.pre_bind_data_id") + + pair_local_ids = [local_id for local_id, _ in prebind_pairs] + pair_remote_ids = [remote_id for _, remote_id in prebind_pairs] + + effective_local_filters = local_filters or list(dict.fromkeys(pair_local_ids)) + effective_remote_filters = remote_filters or list(dict.fromkeys(pair_remote_ids)) + + if set(effective_local_filters) != set(pair_local_ids): + raise ValueError( + "implicit multi-project config mismatch: local project data_id_filter must match project pre_bind_data_id local_data_id" + ) + if set(effective_remote_filters) != set(pair_remote_ids): + raise ValueError( + "implicit multi-project config mismatch: remote project data_id_filter must match project pre_bind_data_id remote_data_id" + ) + + seen_local_ids: set[str] = set() + seen_remote_ids: set[str] = set() + items: list[ImplicitProjectItem] = [] + for local_id, remote_id in prebind_pairs: + if local_id in seen_local_ids: + raise ValueError(f"duplicate local project id in implicit multi-project config: {local_id}") + if remote_id in seen_remote_ids: + raise ValueError(f"duplicate remote project id in implicit multi-project config: {remote_id}") + seen_local_ids.add(local_id) + seen_remote_ids.add(remote_id) + items.append(ImplicitProjectItem(local_project_id=local_id, remote_project_id=remote_id)) + return items + + +def _split_implicit_project_node_types(node_types: list[str]) -> tuple[list[str], list[str]]: + if "project" not in node_types: + raise ValueError("implicit multi-project config requires node_types to include 'project'") + project_index = node_types.index("project") + return node_types[:project_index], node_types[project_index:] + + +def _build_implicit_project_config( + base_config: PipelineRunConfig, + *, + item: ImplicitProjectItem, + project_node_types: list[str], +) -> PipelineRunConfig: + return apply_config_overrides( + base_config, + { + "node_types": project_node_types, + "persist": {"wipe_on_start": False}, + "local_datasource": { + "handler_configs": { + "project": {"data_id_filter": [item.local_project_id]}, + } + }, + "remote_datasource": { + "handler_configs": { + "project": {"data_id_filter": [item.remote_project_id]}, + } + }, + "strategies": { + "project": { + "config": { + "pre_bind_data_id": [ + { + "local_data_id": item.local_project_id, + "remote_data_id": item.remote_project_id, + } + ] + } + } + }, + }, + ) + + +async def run_implicit_project_batch_from_config( + config: PipelineRunConfig, + *, + title: str, + print_summary: bool = True, + max_concurrency: int | None = None, +) -> Dict[str, Dict[str, object]]: + from .factory import create_pipeline_from_config, create_scope_runtime_from_config + + project_items = _extract_implicit_project_items(config) + shared_node_types, project_node_types = _split_implicit_project_node_types(list(config.node_types)) + results: Dict[str, Dict[str, object]] = {} + + global_runtime = None + global_pipeline = None + global_config = None + + if print_summary: + logger.info("\n%s", "=" * 80) + logger.info("🚀 Creating implicit multi-project pipeline from in-memory config") + logger.info("📦 Projects=%d", len(project_items)) + logger.info("=" * 80) + + try: + if shared_node_types: + global_config = apply_config_overrides( + config, + { + "node_types": shared_node_types, + }, + ) + global_runtime = await create_scope_runtime_from_config(global_config, scope="global") + global_pipeline = await create_pipeline_from_config( + global_config, + scope="global", + pipeline_name=f"{title} [global]", + scope_runtime=global_runtime, + ) + results["global"] = await global_pipeline.run() + + semaphore = asyncio.Semaphore(max(1, max_concurrency or len(project_items))) + + async def _run_project(item: ImplicitProjectItem) -> None: + project_config = _build_implicit_project_config( + config, + item=item, + project_node_types=project_node_types, + ) + async with semaphore: + project_runtime = await create_scope_runtime_from_config( + project_config, + scope=item.scope, + persistence_service=global_runtime.persistence_service if global_runtime is not None else None, + local_fallback_collection=global_runtime.local_collection if global_runtime is not None else None, + remote_fallback_collection=global_runtime.remote_collection if global_runtime is not None else None, + binding_fallback_manager=global_runtime.binding_manager if global_runtime is not None else None, + global_runtime=global_runtime, + ) + try: + project_pipeline = await create_pipeline_from_config( + project_config, + scope=item.scope, + pipeline_name=f"{title} [{item.scope}]", + bootstrap_binding_node_types=shared_node_types, + ephemeral_node_types=shared_node_types, + scope_runtime=project_runtime, + ) + results[item.scope] = await project_pipeline.run() + finally: + await project_runtime.unload() + await project_runtime.close() + + await asyncio.gather(*[_run_project(item) for item in project_items]) + return results + finally: + if global_runtime is not None: + await global_runtime.unload() + await global_runtime.close() + + +class ProjectBatchPipeline: def __init__(self, *, project_root: Path, config_path: Path, config: MultiProjectRunConfig): self.project_root = project_root self.config_path = config_path @@ -87,27 +301,6 @@ class MultiProjectPipeline: cfg = apply_config_overrides(cfg, {"persist": {**shared_persist, "wipe_on_start": False}}) return apply_config_overrides(cfg, _project_scope_overrides(item, node_types=list(cfg.node_types))) - async def _snapshot_shared_state(self, *, pipeline, node_types: List[str]) -> _SharedStateSnapshot: - local_nodes = [ - copy.deepcopy(node) - for node_type in node_types - for node in pipeline.local_collection.filter(node_type=node_type) - ] - remote_nodes = [ - copy.deepcopy(node) - for node_type in node_types - for node in pipeline.remote_collection.filter(node_type=node_type) - ] - bindings_by_type: Dict[str, Dict[str, Optional[str]]] = {} - for node_type in node_types: - bindings = await pipeline.binding_manager.get_all_bindings(node_type) - bindings_by_type[node_type] = {local_id: remote_id for local_id, remote_id in bindings} - return { - "local_nodes": local_nodes, - "remote_nodes": remote_nodes, - "bindings_by_type": bindings_by_type, - } - async def run(self) -> Dict[str, Dict[str, object]]: results: Dict[str, Dict[str, object]] = {} @@ -119,49 +312,67 @@ class MultiProjectPipeline: shared_node_types=shared_node_types, ) logger.info("🚀 Multi-project pipeline start: projects=%d", len(self.config.projects)) - global_pipeline = await create_pipeline_from_config( + global_runtime = await create_scope_runtime_from_config( global_config, scope="global", - pipeline_name="multi-project pipeline [global]", ) + global_pipeline = None try: + global_pipeline = await create_pipeline_from_config( + global_config, + scope="global", + pipeline_name="multi-project pipeline [global]", + scope_runtime=global_runtime, + ) results["global"] = await global_pipeline.run() - shared_state = await self._snapshot_shared_state( - pipeline=global_pipeline, - node_types=project_bootstrap_node_types, - ) - finally: - await global_pipeline.close() + semaphore = asyncio.Semaphore(self.config.max_concurrency) - for item in self.config.projects: - project_config = self._prepare_project_config( - shared_persist=shared_persist, - item=item, - base_path=item.config_path, - ) - scope = item.scope - title = f"multi-project pipeline [{scope}]" - logger.info( - "🚀 Project pipeline start: scope=%s local_project_id=%s remote_project_id=%s", - scope, - item.local_project_id, - item.remote_project_id, - ) - project_pipeline = await create_pipeline_from_config( - project_config, - scope=scope, - pipeline_name=title, - bootstrap_binding_node_types=project_bootstrap_node_types, - ephemeral_node_types=project_bootstrap_node_types, - ) - try: - project_pipeline.preload_shared_state( - local_nodes=shared_state["local_nodes"], - remote_nodes=shared_state["remote_nodes"], - bindings_by_type=shared_state["bindings_by_type"], + async def _run_project(item: MultiProjectItemConfig) -> None: + project_config = self._prepare_project_config( + shared_persist=shared_persist, + item=item, + base_path=item.config_path, ) - results[scope] = await project_pipeline.run() - finally: - await project_pipeline.close() + scope = item.scope + title = f"multi-project pipeline [{scope}]" + logger.info( + "🚀 Project pipeline start: scope=%s local_project_id=%s remote_project_id=%s", + scope, + item.local_project_id, + item.remote_project_id, + ) + async with semaphore: + project_runtime = await create_scope_runtime_from_config( + project_config, + scope=scope, + persistence_service=global_runtime.persistence_service, + local_fallback_collection=global_runtime.local_collection, + remote_fallback_collection=global_runtime.remote_collection, + binding_fallback_manager=global_runtime.binding_manager, + global_runtime=global_runtime, + ) + project_pipeline = None + try: + project_pipeline = await create_pipeline_from_config( + project_config, + scope=scope, + pipeline_name=title, + bootstrap_binding_node_types=project_bootstrap_node_types, + ephemeral_node_types=project_bootstrap_node_types, + scope_runtime=project_runtime, + ) + results[scope] = await project_pipeline.run() + finally: + await project_runtime.unload() + await project_runtime.close() - return results \ No newline at end of file + await asyncio.gather(*[_run_project(item) for item in self.config.projects]) + finally: + await global_runtime.unload() + await global_runtime.close() + + return results + + +class MultiProjectPipeline(ProjectBatchPipeline): + pass \ No newline at end of file diff --git a/sync_state_machine/pipeline/project_scope_runtime.py b/sync_state_machine/pipeline/project_scope_runtime.py new file mode 100644 index 0000000..40b5800 --- /dev/null +++ b/sync_state_machine/pipeline/project_scope_runtime.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import asyncio +from typing import List, Optional + +from ..common.binding import BindingManager +from ..common.collection import DataCollection +from ..common.persistence_service import PersistenceService, ScopedPersistenceView + + +class ProjectScopeRuntime: + """Thin runtime container for one scope execution.""" + + def __init__( + self, + *, + scope: str, + persistence_service: PersistenceService, + persistence_view: ScopedPersistenceView, + local_collection: DataCollection, + remote_collection: DataCollection, + binding_manager: BindingManager, + local_datasource, + remote_datasource, + global_runtime: Optional["ProjectScopeRuntime"] = None, + owns_persistence_service: bool = False, + ) -> None: + self.scope = scope + self.persistence_service = persistence_service + self.persistence_view = persistence_view + self.local_collection = local_collection + self.remote_collection = remote_collection + self.binding_manager = binding_manager + self.local_datasource = local_datasource + self.remote_datasource = remote_datasource + self.global_runtime = global_runtime + self.owns_persistence_service = owns_persistence_service + self._closed = False + + @property + def persistence_backend(self): + return self.persistence_service.backend + + async def load( + self, + *, + node_types: List[str], + bootstrap_binding_node_types: Optional[List[str]] = None, + enable_persistence: bool = True, + ephemeral_node_types: Optional[List[str]] = None, + ) -> None: + if not enable_persistence: + return + + excluded_node_types = list(dict.fromkeys(ephemeral_node_types or [])) or None + await self.local_collection.load_from_persistence_excluding(excluded_node_types) + await self.remote_collection.load_from_persistence_excluding(excluded_node_types) + + excluded = set(ephemeral_node_types or []) + binding_node_types = [ + node_type + for node_type in dict.fromkeys([*node_types, *(bootstrap_binding_node_types or [])]) + if node_type not in excluded + ] + for node_type in binding_node_types: + await self.binding_manager.load_from_persistence(node_type) + + async def persist( + self, + *, + enable_persistence: bool = True, + ephemeral_node_types: Optional[List[str]] = None, + ) -> None: + if not enable_persistence: + return + + await self.local_collection.persist(exclude_node_types=ephemeral_node_types) + await self.remote_collection.persist(exclude_node_types=ephemeral_node_types) + await self.binding_manager.persist(exclude_node_types=ephemeral_node_types) + + async def unload(self) -> None: + await self.local_collection.unload() + await self.remote_collection.unload() + await self.binding_manager.unload() + + async def close(self) -> None: + if self._closed: + return + self._closed = True + + await self._safe_close("remote_datasource", self.remote_datasource.close()) + await self._safe_close("local_datasource", self.local_datasource.close()) + if self.owns_persistence_service: + await self._safe_close("persistence_service", self.persistence_service.close()) + + async def _safe_close(self, name: str, close_coro, timeout: float = 3.0) -> None: + try: + await asyncio.wait_for(close_coro, timeout=timeout) + except asyncio.TimeoutError: + return + except Exception: + return \ No newline at end of file diff --git a/tests/unit/test_multi_project_pipeline_config.py b/tests/unit/test_multi_project_pipeline_config.py index e88fe71..ab248fa 100644 --- a/tests/unit/test_multi_project_pipeline_config.py +++ b/tests/unit/test_multi_project_pipeline_config.py @@ -7,14 +7,21 @@ import yaml from pydantic import BaseModel from sync_state_machine.config.multi_project_config import build_multi_project_config_from_file +from sync_state_machine.config import PipelineRunConfig, JsonlDataSourceConfig, PersistConfig, LoggingConfig, StrategyRuntimeConfig, StrategyConfig from sync_state_machine.common.binding import BindingManager from sync_state_machine.common.collection import DataCollection from sync_state_machine.common.sqlite_backend import SQLitePersistenceBackend +from sync_state_machine.common.persistence_service import PersistenceService from sync_state_machine.common.sync_node import SyncNode +from sync_state_machine.domain.project.sync_node import ProjectSyncNode from sync_state_machine.pipeline.multi_project_pipeline import ( + _extract_implicit_project_items, + _split_implicit_project_node_types, + is_implicit_project_batch_config, _project_scope_overrides, _resolve_project_bootstrap_node_types, ) +from sync_state_machine.pipeline.factory import run_pipeline_from_config def test_build_multi_project_config_reads_explicit_project_bootstrap_node_types(tmp_path: Path) -> None: @@ -43,6 +50,32 @@ def test_build_multi_project_config_reads_explicit_project_bootstrap_node_types( assert config.project_bootstrap_node_types == ["company", "supplier"] +def test_build_multi_project_config_reads_max_concurrency(tmp_path: Path) -> None: + config_path = tmp_path / "multi.yaml" + config_path.write_text( + yaml.safe_dump( + { + "global_config": "run_profiles/global.yaml", + "default_project_config": "run_profiles/project.yaml", + "max_concurrency": 8, + "projects": [ + { + "local_project_id": "local-project", + "remote_project_id": "remote-project", + } + ], + }, + allow_unicode=True, + sort_keys=False, + ), + encoding="utf-8", + ) + + config = build_multi_project_config_from_file(project_root=tmp_path, file_path=config_path) + + assert config.max_concurrency == 8 + + def test_project_scope_overrides_only_inject_project_handler_filter() -> None: item = type("Item", (), {"local_project_id": "local-project", "remote_project_id": "remote-project"})() @@ -120,4 +153,218 @@ async def test_binding_manager_prunes_stale_bindings_after_project_scope_cleanup assert await binding_manager.get_remote_id("project", "local-project-1") == "remote-project-1" assert await binding_manager.get_remote_id("project", "local-project-2") is None finally: - await persistence.close() \ No newline at end of file + await persistence.close() + + +@pytest.mark.asyncio +async def test_collection_uses_scoped_persistence_view_and_global_fallback() -> None: + backend = SQLitePersistenceBackend(":memory:") + service = PersistenceService(backend) + await service.initialize() + try: + global_view = service.view("global") + project_view = service.view("project_id:p1") + + global_collection = DataCollection("local", scope="global", persistence=global_view, auto_persist=False) + project_collection = DataCollection( + "local", + scope="project_id:p1", + persistence=project_view, + auto_persist=False, + global_fallback_collection=global_collection, + ) + + global_node = ProjectSyncNode(node_id="global-project-1", data={"id": "p1", "name": "Project P1"}, data_id="p1") + await global_collection.add(global_node) + await global_collection.persist() + await global_collection.unload() + await global_collection.load_from_persistence() + + assert project_collection.get_by_data_id("project", "p1").node_id == "global-project-1" + + project_node = ProjectSyncNode(node_id="project-local-1", data={"id": "p1", "name": "Project P1 Local"}, data_id="p1") + await project_collection.add(project_node) + + assert project_collection.get_by_data_id("project", "p1").node_id == "project-local-1" + + await project_collection.unload() + assert project_collection.filter() == [] + assert project_collection.get_by_data_id("project", "p1").node_id == "global-project-1" + finally: + await service.close() + + +@pytest.mark.asyncio +async def test_binding_manager_uses_scoped_persistence_view_and_global_fallback() -> None: + backend = SQLitePersistenceBackend(":memory:") + service = PersistenceService(backend) + await service.initialize() + try: + global_view = service.view("global") + project_view = service.view("project_id:p1") + + global_binding_manager = BindingManager(global_view, auto_persist=False, scope="global") + project_binding_manager = BindingManager( + project_view, + auto_persist=False, + scope="project_id:p1", + global_fallback_manager=global_binding_manager, + ) + + await global_binding_manager.bind("project", "local-global-1", "remote-global-1") + await global_binding_manager.persist() + await global_binding_manager.unload() + await global_binding_manager.load_from_persistence("project") + + assert await project_binding_manager.get_remote_id("project", "local-global-1") == "remote-global-1" + assert await project_binding_manager.get_local_id("project", "remote-global-1") == "local-global-1" + + await project_binding_manager.bind("project", "local-global-1", "remote-project-1") + assert await project_binding_manager.get_remote_id("project", "local-global-1") == "remote-project-1" + + await project_binding_manager.unload() + assert await project_binding_manager.get_remote_id("project", "local-global-1") == "remote-global-1" + finally: + await service.close() + + +def test_regular_pipeline_config_can_be_detected_as_implicit_multi_project() -> None: + config = PipelineRunConfig( + node_types=["supplier", "company", "project", "contract"], + local_datasource=JsonlDataSourceConfig( + type="jsonl", + jsonl_dir="./local", + handler_configs={ + "project": { + "data_id_filter": [ + "019a62a5-2323-7be2-99c8-82534f8befa4", + "019c27e4-8b34-7779-9ddb-b8b84913e9bb", + ] + } + }, + ), + remote_datasource=JsonlDataSourceConfig( + type="jsonl", + jsonl_dir="./remote", + handler_configs={ + "project": { + "data_id_filter": [ + "7119be34-57f2-4023-a251-dab30a0e6d7b", + "403a2a00-d983-4327-a52f-997bf129b833", + ] + } + }, + ), + persist=PersistConfig(backend="sqlite", db_path=":memory:"), + logging=LoggingConfig(), + strategies=[ + StrategyRuntimeConfig( + node_type="project", + config=StrategyConfig( + auto_bind=False, + pre_bind_data_id=[ + { + "local_data_id": "019a62a5-2323-7be2-99c8-82534f8befa4", + "remote_data_id": "7119be34-57f2-4023-a251-dab30a0e6d7b", + }, + { + "local_data_id": "019c27e4-8b34-7779-9ddb-b8b84913e9bb", + "remote_data_id": "403a2a00-d983-4327-a52f-997bf129b833", + }, + ], + ), + ) + ], + ) + + assert is_implicit_project_batch_config(config) is True + assert _split_implicit_project_node_types(list(config.node_types)) == ( + ["supplier", "company"], + ["project", "contract"], + ) + assert _extract_implicit_project_items(config) == [ + type(_extract_implicit_project_items(config)[0])( + local_project_id="019a62a5-2323-7be2-99c8-82534f8befa4", + remote_project_id="7119be34-57f2-4023-a251-dab30a0e6d7b", + ), + type(_extract_implicit_project_items(config)[0])( + local_project_id="019c27e4-8b34-7779-9ddb-b8b84913e9bb", + remote_project_id="403a2a00-d983-4327-a52f-997bf129b833", + ), + ] + + +@pytest.mark.asyncio +async def test_run_pipeline_from_config_can_force_full_pipeline(monkeypatch: pytest.MonkeyPatch) -> None: + config = PipelineRunConfig( + node_types=["project"], + local_datasource=JsonlDataSourceConfig( + type="jsonl", + jsonl_dir="./local", + handler_configs={"project": {"data_id_filter": ["p1", "p2"]}}, + ), + remote_datasource=JsonlDataSourceConfig( + type="jsonl", + jsonl_dir="./remote", + handler_configs={"project": {"data_id_filter": ["r1", "r2"]}}, + ), + persist=PersistConfig(backend="sqlite", db_path=":memory:"), + logging=LoggingConfig(), + strategies=[ + StrategyRuntimeConfig( + node_type="project", + config=StrategyConfig( + auto_bind=False, + pre_bind_data_id=[ + {"local_data_id": "p1", "remote_data_id": "r1"}, + {"local_data_id": "p2", "remote_data_id": "r2"}, + ], + ), + ) + ], + ) + + captured: dict[str, object] = {} + + class _StubPipeline: + async def run(self): + captured["ran_full"] = True + return {"mode": "full"} + + async def close(self): + captured["closed_full"] = True + + async def fake_create_pipeline_from_config(*args, **kwargs): + captured["scope"] = kwargs.get("scope") + return _StubPipeline() + + async def fake_run_implicit_project_batch_from_config(*args, **kwargs): + captured["ran_multi"] = True + return {"mode": "multi-project"} + + monkeypatch.setattr("sync_state_machine.pipeline.factory.create_pipeline_from_config", fake_create_pipeline_from_config) + monkeypatch.setattr( + "sync_state_machine.pipeline.multi_project_pipeline.run_implicit_project_batch_from_config", + fake_run_implicit_project_batch_from_config, + ) + + result = await run_pipeline_from_config(config, print_summary=False, pipeline_type="full") + + assert result == {"mode": "full"} + assert captured["ran_full"] is True + assert "ran_multi" not in captured + + +@pytest.mark.asyncio +async def test_run_pipeline_from_config_can_require_multi_project(monkeypatch: pytest.MonkeyPatch) -> None: + config = PipelineRunConfig( + node_types=["project"], + local_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir="./local"), + remote_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir="./remote"), + persist=PersistConfig(backend="sqlite", db_path=":memory:"), + logging=LoggingConfig(), + strategies=[], + ) + + with pytest.raises(ValueError, match="pipeline_type=multi-project"): + await run_pipeline_from_config(config, print_summary=False, pipeline_type="multi-project") \ No newline at end of file diff --git a/tests/unit/test_summary_report.py b/tests/unit/test_summary_report.py index d175e6e..b2946a3 100644 --- a/tests/unit/test_summary_report.py +++ b/tests/unit/test_summary_report.py @@ -6,6 +6,7 @@ from io import StringIO import pytest from sync_state_machine.pipeline.summary_report import print_pipeline_summary +from sync_state_machine.logging import get_scope_logger class _FakeBindingManager: @@ -75,4 +76,24 @@ async def test_pipeline_summary_omits_diff_schemas_column() -> None: assert "ProjectBaseInfoUpdate" not in output assert "field mismatch/total samples" in output assert "actual_production_date" in output - assert "None/'2025-11-14'" in output \ No newline at end of file + assert "None/'2025-11-14'" in output + + +def test_scope_logger_prefixes_messages() -> None: + stream = StringIO() + base_logger = logging.getLogger("sync_state_machine.tests.scope_logger") + base_logger.handlers.clear() + base_logger.setLevel(logging.INFO) + base_logger.propagate = False + + handler = logging.StreamHandler(stream) + handler.setFormatter(logging.Formatter("%(message)s")) + base_logger.addHandler(handler) + + try: + scoped_logger = get_scope_logger("project_id:p1", "tests.scope_logger") + scoped_logger.info("Strategy start: project") + finally: + base_logger.removeHandler(handler) + + assert stream.getvalue().strip() == "[project_id:p1] Strategy start: project" \ No newline at end of file