strepsiades/ecm_sync_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

多pipeline同步支持

Browse Source
This commit is contained in:
strepsiades
2026-03-23 21:02:52 +08:00
parent 0812489f9d
commit b51ed0175b
49 changed files with 2110 additions and 792 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sync_state_machine/pipeline/__init__.py
+4
View File
@@ -4,6 +4,7 @@ Pipeline Layer - Pipeline orchestration and workflow management
from .copy_data_pipeline import CopyDataPipeline
from .full_sync_pipeline import FullSyncPipeline
from .multi_project_pipeline import MultiProjectPipeline
from ..config import (
PipelineRunConfig,
ApiDataSourceConfig,
@@ -24,11 +25,13 @@ from .factory import (
create_pipeline_from_config,
run_pipeline_from_file,
run_pipeline_from_config,
run_profile_from_file,
)
__all__ = [
"CopyDataPipeline",
"FullSyncPipeline",
"MultiProjectPipeline",
"create_jsonl_to_api_pipeline",
"create_jsonl_to_jsonl_pipeline",
"create_api_to_jsonl_pipeline",
@@ -36,6 +39,7 @@ __all__ = [
"create_pipeline_from_config",
"run_pipeline_from_file",
"run_pipeline_from_config",
"run_profile_from_file",
"PipelineRunConfig",
"ApiDataSourceConfig",
"JsonlDataSourceConfig",
sync_state_machine/pipeline/factory.py
+111 -24
View File
@@ -27,18 +27,23 @@ from ..datasource import ensure_builtin_datasource_types_registered
from ..datasource.type_registry import get_datasource_factory
from ..config import (
PipelineRunConfig,
MultiProjectRunConfig,
DataSourceConfig,
ApiDataSourceConfig,
PersistConfig,
JsonlDataSourceConfig,
LoggingConfig,
build_config,
StrategyRuntimeConfig,
build_config_from_file,
build_multi_project_config_from_file,
apply_config_overrides,
OrphanAction,
UpdateDirection,
resolve_log_file_path,
config_snapshot,
is_multi_project_payload,
load_overrides_from_file,
)
from ..sync_system.strategy import BaseSyncStrategy
from .run_logger import init_pipeline_logger
@@ -180,6 +185,17 @@ def _register_handlers(
)
def _build_collections_and_bindings(
*,
persistence: PersistenceBackend,
scope: str,
) -> 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)
return local_collection, remote_collection, binding_manager
def _resolve_strategy_map(config: PipelineRunConfig, node_types: List[str]) -> Dict[str, StrategyRuntimeConfig]:
if config.strategies:
return {item.node_type: item for item in config.strategies}
@@ -196,6 +212,28 @@ def _resolve_strategy_map(config: PipelineRunConfig, node_types: List[str]) -> D
return resolved
def _build_pipeline_run_config(
*,
node_types: List[str],
local_datasource: DataSourceConfig,
remote_datasource: DataSourceConfig,
persist: PersistConfig,
logging_config: LoggingConfig,
strategy_overrides: Optional[Dict[str, Dict[str, Any]]] = None,
) -> PipelineRunConfig:
cfg = PipelineRunConfig(
node_types=node_types,
local_datasource=local_datasource,
remote_datasource=remote_datasource,
strategies=[],
persist=persist,
logging=logging_config,
)
if strategy_overrides:
cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides})
return cfg
def _build_strategies(
config: PipelineRunConfig,
node_types: List[str],
@@ -225,6 +263,7 @@ def _build_strategies(
runtime_cfg = strategy_map.get(node_type)
if runtime_cfg is not None:
strategy.set_config(runtime_cfg.config.model_copy(deep=True))
strategy.skip_sync = bool(runtime_cfg.skip_sync)
strategies.append(strategy)
return strategies
@@ -233,6 +272,10 @@ def _build_strategies(
async def create_pipeline_from_config(
config: PipelineRunConfig,
*,
scope: str = "global",
pipeline_name: str | None = None,
bootstrap_binding_node_types: Optional[List[str]] = None,
ephemeral_node_types: Optional[List[str]] = None,
local_datasource_override=None,
remote_datasource_override=None,
) -> FullSyncPipeline:
@@ -279,9 +322,10 @@ async def create_pipeline_from_config(
await _initialize_datasource(remote_ds, endpoint_name="remote")
_register_handlers(config, local_ds, remote_ds, all_types)
local_collection = DataCollection("local", persistence, auto_persist=False)
remote_collection = DataCollection("remote", persistence, auto_persist=False)
binding_manager = BindingManager(persistence, auto_persist=False)
local_collection, remote_collection, binding_manager = _build_collections_and_bindings(
persistence=persistence,
scope=scope,
)
strategies = _build_strategies(
config,
@@ -301,6 +345,10 @@ async def create_pipeline_from_config(
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,
)
@@ -325,6 +373,8 @@ async def run_pipeline_from_config(
*,
title: str = "sync_state_machine pipeline",
print_summary: bool = True,
scope: str = "global",
bootstrap_binding_node_types: Optional[List[str]] = None,
local_datasource_override=None,
remote_datasource_override=None,
) -> Dict[str, Any]:
@@ -341,7 +391,7 @@ async def run_pipeline_from_config(
resolved_cfg = config_snapshot(config)
resolved_cfg_text = json.dumps(resolved_cfg, ensure_ascii=False, indent=2)
logger.info(
"Resolved Config Summary: node_types=%d local=%s remote=%s",
"Resolved Full Config Summary: node_types=%d local=%s remote=%s",
len(config.node_types),
config.local_datasource.type,
config.remote_datasource.type,
@@ -349,7 +399,7 @@ async def run_pipeline_from_config(
if config.logging.file_path:
try:
with open(config.logging.file_path, "a", encoding="utf-8") as fp:
fp.write("Resolved Config:\n")
fp.write("Resolved Full Config:\n")
fp.write(resolved_cfg_text)
fp.write("\n")
except Exception as exc:
@@ -362,15 +412,18 @@ async def run_pipeline_from_config(
print(f"🚀 Creating {title}")
print(f"📝 Log file: {config.logging.file_path or '-'}")
print(
f"🔧 Resolved Config Summary: node_types={len(config.node_types)} "
f"🔧 Resolved Full Config Summary: node_types={len(config.node_types)} "
f"local={config.local_datasource.type} "
f"remote={config.remote_datasource.type}"
)
print(f"🔧 Resolved Config:\n{resolved_cfg_text}")
print(f"🔧 Resolved Full Config:\n{resolved_cfg_text}")
print("=" * 80)
pipeline = await create_pipeline_from_config(
config,
scope=scope,
pipeline_name=title,
bootstrap_binding_node_types=bootstrap_binding_node_types,
local_datasource_override=local_datasource_override,
remote_datasource_override=remote_datasource_override,
)
@@ -390,12 +443,20 @@ async def create_pipeline_from_file(
*,
project_root: Path,
config_path: str | Path,
scope: str = "global",
pipeline_name: str | None = None,
bootstrap_binding_node_types: Optional[List[str]] = None,
ephemeral_node_types: Optional[List[str]] = None,
local_datasource_override=None,
remote_datasource_override=None,
) -> FullSyncPipeline:
config = build_config_from_file(project_root=project_root, file_path=config_path)
return await create_pipeline_from_config(
config,
scope=scope,
pipeline_name=pipeline_name,
bootstrap_binding_node_types=bootstrap_binding_node_types,
ephemeral_node_types=ephemeral_node_types,
local_datasource_override=local_datasource_override,
remote_datasource_override=remote_datasource_override,
)
@@ -407,6 +468,8 @@ async def run_pipeline_from_file(
config_path: str | Path,
title: str = "sync_state_machine pipeline",
print_summary: bool = True,
scope: str = "global",
bootstrap_binding_node_types: Optional[List[str]] = None,
local_datasource_override=None,
remote_datasource_override=None,
) -> Dict[str, Any]:
@@ -415,11 +478,41 @@ async def run_pipeline_from_file(
config,
title=title,
print_summary=print_summary,
scope=scope,
bootstrap_binding_node_types=bootstrap_binding_node_types,
local_datasource_override=local_datasource_override,
remote_datasource_override=remote_datasource_override,
)
async def run_profile_from_file(
*,
project_root: Path,
config_path: str | Path,
print_summary: bool = True,
) -> Dict[str, Any]:
cfg_path = Path(config_path)
raw = load_overrides_from_file(cfg_path, project_root)
if is_multi_project_payload(raw):
from .multi_project_pipeline import MultiProjectPipeline
multi_cfg = MultiProjectRunConfig.model_validate(raw)
pipeline = MultiProjectPipeline(project_root=project_root, config_path=cfg_path, config=multi_cfg)
if print_summary:
print("\n" + "=" * 80)
print(f"🚀 Creating multi-project pipeline from: {cfg_path}")
print("=" * 80)
return await pipeline.run()
config = build_config(project_root=project_root, overrides=raw)
mode = f"{config.local_datasource.type}->{config.remote_datasource.type}"
return await run_pipeline_from_config(
config,
title=f"sync_state_machine pipeline ({mode})",
print_summary=print_summary,
)
async def create_jsonl_to_api_pipeline(
*,
# 路径配置
@@ -438,7 +531,7 @@ async def create_jsonl_to_api_pipeline(
# API 配置
api_debug: bool = False,
poll_max_retries: int = 10,
poll_interval: float = 0.5,
poll_interval: float = 0.5,
api_rate_limit_max_requests: int = 30,
api_rate_limit_window_seconds: float = 10.0,
local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None,
@@ -448,7 +541,7 @@ async def create_jsonl_to_api_pipeline(
logger_file_path: Optional[str] = None,
logger_level: int = logging.INFO,
) -> FullSyncPipeline:
cfg = PipelineRunConfig(
cfg = _build_pipeline_run_config(
node_types=node_types,
local_datasource=JsonlDataSourceConfig(
type="jsonl",
@@ -468,20 +561,18 @@ async def create_jsonl_to_api_pipeline(
api_rate_limit_window_seconds=api_rate_limit_window_seconds,
handler_configs=remote_handler_configs or {},
),
strategies=[],
persist=PersistConfig(
db_path=persistence_db,
enable=enable_persistence,
wipe_on_start=wipe_persistence_on_start,
),
logging=LoggingConfig(
logging_config=LoggingConfig(
initialize=initialize_logger,
file_path=logger_file_path,
level=logger_level,
),
strategy_overrides=strategy_overrides,
)
if strategy_overrides:
cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides})
return await create_pipeline_from_config(cfg)
@@ -503,7 +594,7 @@ async def create_api_to_jsonl_pipeline(
# API 配置
api_debug: bool = False,
poll_max_retries: int = 10,
poll_interval: float = 0.5,
poll_interval: float = 0.5,
api_rate_limit_max_requests: int = 30,
api_rate_limit_window_seconds: float = 10.0,
local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None,
@@ -513,7 +604,7 @@ async def create_api_to_jsonl_pipeline(
logger_file_path: Optional[str] = None,
logger_level: int = logging.INFO,
) -> FullSyncPipeline:
cfg = PipelineRunConfig(
cfg = _build_pipeline_run_config(
node_types=node_types,
local_datasource=ApiDataSourceConfig(
type="api",
@@ -533,20 +624,18 @@ async def create_api_to_jsonl_pipeline(
read_only=False,
handler_configs=remote_handler_configs or {},
),
strategies=[],
persist=PersistConfig(
db_path=persistence_db,
enable=enable_persistence,
wipe_on_start=wipe_persistence_on_start,
),
logging=LoggingConfig(
logging_config=LoggingConfig(
initialize=initialize_logger,
file_path=logger_file_path,
level=logger_level,
),
strategy_overrides=strategy_overrides,
)
if strategy_overrides:
cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides})
return await create_pipeline_from_config(cfg)
@@ -565,7 +654,7 @@ async def create_jsonl_to_jsonl_pipeline(
logger_file_path: Optional[str] = None,
logger_level: int = logging.INFO,
) -> FullSyncPipeline:
cfg = PipelineRunConfig(
cfg = _build_pipeline_run_config(
node_types=node_types,
local_datasource=JsonlDataSourceConfig(
type="jsonl",
@@ -579,18 +668,16 @@ async def create_jsonl_to_jsonl_pipeline(
read_only=False,
handler_configs=remote_handler_configs or {},
),
strategies=[],
persist=PersistConfig(
db_path=persistence_db,
enable=enable_persistence,
wipe_on_start=wipe_persistence_on_start,
),
logging=LoggingConfig(
logging_config=LoggingConfig(
initialize=initialize_logger,
file_path=logger_file_path,
level=logger_level,
),
strategy_overrides=strategy_overrides,
)
if strategy_overrides:
cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides})
return await create_pipeline_from_config(cfg)
sync_state_machine/pipeline/full_sync_pipeline.py
+105 -26
View File
@@ -12,6 +12,7 @@ from pathlib import Path
if TYPE_CHECKING:
from ..datasource.datasource import BaseDataSource
from ..datasource.jsonl import JsonlDataSource
from ..common.collection import DataCollection
from ..common.binding import BindingManager
from ..common.persistence import PersistenceBackend
@@ -47,6 +48,10 @@ class FullSyncPipeline:
local_collection: DataCollection,
remote_collection: DataCollection,
binding_manager: BindingManager,
scope: str = "global",
pipeline_name: str | None = None,
bootstrap_binding_node_types: Optional[List[str]] = None,
ephemeral_node_types: Optional[List[str]] = None,
node_types: Optional[List[str]] = None,
load_order: Optional[List[str]] = None,
sync_order: Optional[List[str]] = None,
@@ -61,7 +66,11 @@ class FullSyncPipeline:
self.local_datasource = local_datasource
self.remote_datasource = remote_datasource
self.strategies = strategies
self.scope = scope
self.pipeline_name = pipeline_name or f"pipeline[{scope}]"
self.node_types = node_types or [s.node_type for s in self.strategies]
self.bootstrap_binding_node_types = list(dict.fromkeys(bootstrap_binding_node_types or []))
self.ephemeral_node_types = list(dict.fromkeys(ephemeral_node_types or []))
self.load_order = load_order or self.node_types
self.sync_order = sync_order or self.node_types
self.persistence = persistence
@@ -91,6 +100,9 @@ class FullSyncPipeline:
self._logger = logger
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:
@@ -104,17 +116,17 @@ class FullSyncPipeline:
async def run(self) -> Dict[str, Any]:
try:
await self.phase_bootstrap()
self._logger.info("✅ Phase bootstrap completed")
self._logger.info("✅ Phase bootstrap completed (%s)", self.pipeline_name)
await self.phase_sync_by_node_type()
self._logger.info("✅ Phase sync-by-node-type completed")
self._logger.info("✅ Phase sync-by-node-type completed (%s)", self.pipeline_name)
post_ok = await self.phase_post_check()
self.stats["post_check_passed"] = post_ok
self._logger.info("✅ Phase post-check completed")
self._logger.info("✅ Phase post-check completed (%s)", self.pipeline_name)
await self.phase_persistence()
self._logger.info("✅ Phase persistence completed")
self._logger.info("✅ Phase persistence completed (%s)", self.pipeline_name)
return self.stats
finally:
await self.close()
@@ -173,9 +185,14 @@ class FullSyncPipeline:
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")
await self._apply_project_scope_cleanup()
async def phase_bind(self) -> None:
self._prepare_datasources()
strategy_map = {s.node_type: s for s in self.strategies}
@@ -257,10 +274,17 @@ class FullSyncPipeline:
if not self.enable_persistence:
return
excluded_node_types = self.ephemeral_node_types or None
# 加载持久化数据
await self.local_collection.load_from_persistence()
await self.remote_collection.load_from_persistence()
for node_type in self.node_types:
await self.local_collection.load_from_persistence_excluding(excluded_node_types)
await self.remote_collection.load_from_persistence_excluding(excluded_node_types)
binding_node_types = [
node_type
for node_type in dict.fromkeys([*self.node_types, *self.bootstrap_binding_node_types])
if node_type not in set(self.ephemeral_node_types)
]
for node_type in binding_node_types:
await self.binding_manager.load_from_persistence(node_type)
self._logger.info("🔄 Persistence loaded (state preserved; pending E01 normalization)")
@@ -278,11 +302,67 @@ class FullSyncPipeline:
self._logger.info(f"🧹 Reset cleanup: {total_cleaned} CREATE-failed zombies cleaned")
self._logger.info("🔄 Post-load reset cleanup completed")
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)
self.remote_datasource.set_collection(self.remote_collection)
def _get_project_scope_filter_ids(self, datasource: "BaseDataSource") -> List[str]:
try:
handler = datasource.get_handler("project")
get_data_id_filter = getattr(handler, "get_data_id_filter", None)
if callable(get_data_id_filter):
raw_filter = get_data_id_filter()
if isinstance(raw_filter, list):
return list(dict.fromkeys(raw_filter))
except Exception:
pass
return []
async def _apply_project_scope_cleanup(self) -> None:
local_project_ids = self._get_project_scope_filter_ids(self.local_datasource)
remote_project_ids = self._get_project_scope_filter_ids(self.remote_datasource)
if not local_project_ids and not remote_project_ids:
return
deleted_local = await self.local_collection.filter_by_project_ids(local_project_ids)
deleted_remote = await self.remote_collection.filter_by_project_ids(remote_project_ids)
deleted_bindings = await self.binding_manager.prune_missing_nodes(
local_collection=self.local_collection,
remote_collection=self.remote_collection,
node_types=list(dict.fromkeys([*self.node_types, *self.bootstrap_binding_node_types])),
)
if deleted_local or deleted_remote or deleted_bindings:
self._logger.info(
"🧹 Project-scope cleanup: local_deleted=%d remote_deleted=%d bindings_deleted=%d",
deleted_local,
deleted_remote,
deleted_bindings,
)
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])
@@ -300,7 +380,13 @@ class FullSyncPipeline:
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
def _should_skip_reload(self) -> bool:
return isinstance(self.local_datasource, JsonlDataSource) and isinstance(self.remote_datasource, JsonlDataSource)
async def _reload_node_type(self, node_type: str, *, reason: str) -> None:
if self._should_skip_reload():
self._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", node_type, reason)
return
await self._load_node_type(node_type, reason=reason)
async def _evaluate_consistency_for_node_type(self, node_type: str) -> Dict[str, Any]:
@@ -309,28 +395,18 @@ class FullSyncPipeline:
compare_to_remote = True
if strategy and strategy.config.update_direction == UpdateDirection.PULL:
compare_to_remote = False
ignore_list_item_fields = dict(strategy.config.post_check_ignore_list_item_fields) if strategy else {}
# 从远端 handler 的 update_schemas 提取只读字段过滤白名单
post_check_fields: List[str] | None = None
try:
remote_handler = self.remote_datasource.get_handler(node_type)
fields = remote_handler.get_update_fields()
if fields:
post_check_fields = fields
except Exception:
pass
ignore_fields = list(strategy.config.post_check_ignore_fields) if strategy else []
result = await evaluate_consistency_for_node_type(
node_type=node_type,
strategy=strategy,
local_collection=self.local_collection,
remote_collection=self.remote_collection,
binding_manager=self.binding_manager,
logger=self._logger,
depend_fields=depend_fields,
compare_to_remote=compare_to_remote,
ignore_list_item_fields=ignore_list_item_fields,
post_check_fields=post_check_fields,
ignore_fields=ignore_fields,
)
self._consistency_by_type[node_type] = result
return result
@@ -430,9 +506,12 @@ class FullSyncPipeline:
所有类型均已同步完毕后，统一做一次全量 reload（触发后端派生字段计算），
再对所有类型做一致性评估，确保 post-check 结果反映最终状态。
"""
self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...")
for node_type in self.node_types:
await self._reload_node_type(node_type, reason="post-check final reload")
if self._should_skip_reload():
self._logger.info("⏭️ Post-check reload skipped for JSONL pipeline")
else:
self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...")
for node_type in self.node_types:
await self._reload_node_type(node_type, reason="post-check final reload")
for node_type in self.node_types:
await self._evaluate_consistency_for_node_type(node_type)
@@ -454,11 +533,11 @@ class FullSyncPipeline:
return
self._logger.info("🔍 Persisting local collection...")
await self.local_collection.persist()
await self.local_collection.persist(exclude_node_types=self.ephemeral_node_types)
self._logger.info("🔍 Persisting remote collection...")
await self.remote_collection.persist()
await self.remote_collection.persist(exclude_node_types=self.ephemeral_node_types)
self._logger.info("🔍 Persisting binding manager...")
await self.binding_manager.persist()
await self.binding_manager.persist(exclude_node_types=self.ephemeral_node_types)
self._logger.info("✅ All data persisted successfully")
# ========== Summary & Reporting ==========
sync_state_machine/pipeline/multi_project_pipeline.py
+166
View File
@@ -0,0 +1,166 @@
from __future__ import annotations
import logging
import copy
from pathlib import Path
from typing import Dict, Optional, List, TypedDict, Any
from ..config import (
MultiProjectItemConfig,
MultiProjectRunConfig,
PipelineRunConfig,
apply_config_overrides,
build_config_from_file,
)
from .factory import create_pipeline_from_config
logger = logging.getLogger(__name__)
class _SharedStateSnapshot(TypedDict):
local_nodes: List[Any]
remote_nodes: List[Any]
bindings_by_type: Dict[str, Dict[str, Optional[str]]]
def _resolve_embedded_config_path(raw_path: str, *, project_root: Path, profile_path: Path) -> Path:
candidate = Path(raw_path)
if candidate.is_absolute():
return candidate
profile_relative = (profile_path.parent / candidate).resolve()
if profile_relative.exists():
return profile_relative
return (project_root / candidate).resolve()
def _project_scope_overrides(item: MultiProjectItemConfig, *, node_types: list[str]) -> Dict[str, object]:
del node_types
local_handler_configs = {
"project": {"data_id_filter": [item.local_project_id]},
}
remote_handler_configs = {
"project": {"data_id_filter": [item.remote_project_id]},
}
return {
"local_datasource": {
"handler_configs": local_handler_configs,
},
"remote_datasource": {
"handler_configs": remote_handler_configs,
},
}
def _resolve_project_bootstrap_node_types(
*,
config: MultiProjectRunConfig,
shared_node_types: list[str],
) -> list[str]:
configured = [node_type for node_type in config.project_bootstrap_node_types if node_type]
if configured:
return list(dict.fromkeys(configured))
return list(dict.fromkeys(shared_node_types))
class MultiProjectPipeline:
def __init__(self, *, project_root: Path, config_path: Path, config: MultiProjectRunConfig):
self.project_root = project_root
self.config_path = config_path
self.config = config
def _load_pipeline_config(self, raw_path: str) -> PipelineRunConfig:
resolved = _resolve_embedded_config_path(raw_path, project_root=self.project_root, profile_path=self.config_path)
return build_config_from_file(project_root=self.project_root, file_path=resolved)
def _prepare_project_config(
self,
*,
shared_persist: Dict[str, object],
item: MultiProjectItemConfig,
base_path: Optional[str],
) -> PipelineRunConfig:
raw_path = base_path or self.config.default_project_config
cfg = self._load_pipeline_config(raw_path)
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]] = {}
global_config = self._load_pipeline_config(self.config.global_config)
shared_persist = global_config.persist.model_dump(mode="json")
shared_node_types = list(global_config.node_types)
project_bootstrap_node_types = _resolve_project_bootstrap_node_types(
config=self.config,
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_config,
scope="global",
pipeline_name="multi-project pipeline [global]",
)
try:
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()
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"],
)
results[scope] = await project_pipeline.run()
finally:
await project_pipeline.close()
return results
sync_state_machine/pipeline/summary_report.py
+3 -4
View File
@@ -52,18 +52,17 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d
logger.info(" pass")
continue
logger.info(" field mismatch/total diff_schemas samples")
logger.info(" ------------------------------------------------------------------------------")
logger.info(" field mismatch/total samples")
logger.info(" -----------------------------------------------------------")
for field_report in differing_fields:
field_name = str(field_report.get("field_name", ""))
field_mismatch = int(field_report.get("mismatch_count", 0))
field_total = int(field_report.get("total_count", 0))
schema_refs = ",".join(str(item) for item in (field_report.get("schema_refs", []) or [])) or "-"
samples = field_report.get("samples", []) or []
sample_text = _format_sample(samples[0]) if samples else "-"
logger.info(
f" {_clip(field_name, 30):<30} {field_mismatch:>4}/{field_total:<4} "
f"{_clip(schema_refs, 28):<28} {sample_text}"
f"{sample_text}"
)
def print_collection_summary(title: str, datasource, collection) -> None: