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

整理了初始化过程，现在更容易被继承了。

Browse Source
This commit is contained in:
strepsiades
2026-03-18 16:48:53 +08:00
parent 84b3b7652a
commit 841986740c
22 changed files with 432 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sync_state_machine/__init__.py
+3 -3
View File
@@ -15,10 +15,10 @@ from .common.registry import DomainRegistry
from .config import (
ApiDataSourceConfig,
BaseDataSourceConfig,
DbDataSourceConfig,
JsonlDataSourceConfig,
PipelineRunConfig,
build_config,
build_config_from_file,
build_default_config,
get_datasource_config_model,
get_datasource_factory,
@@ -40,7 +40,7 @@ except PackageNotFoundError:
def ensure_domain_registry_loaded() -> None:
"""Ensure built-in domain registrations are imported."""
_domain # keep module import for side effects
_ = _domain
__all__ = [
@@ -50,9 +50,9 @@ __all__ = [
"PipelineRunConfig",
"ApiDataSourceConfig",
"BaseDataSourceConfig",
"DbDataSourceConfig",
"JsonlDataSourceConfig",
"build_config",
"build_config_from_file",
"build_default_config",
"get_datasource_config_model",
"get_datasource_factory",
sync_state_machine/config/__init__.py
+3 -2
View File
@@ -1,4 +1,4 @@
from .datasource_config import ApiDataSourceConfig, BaseDataSourceConfig, DbDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
from .datasource_config import ApiDataSourceConfig, BaseDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
from .persist_config import PersistConfig
from .logging_config import LoggingConfig, resolve_log_file_path
from ..datasource.type_registry import (
@@ -27,6 +27,7 @@ from .pipeline_config import PipelineRunConfig, DEFAULT_NODE_TYPES
from .builder import (
build_default_config,
build_config,
build_config_from_file,
apply_config_overrides,
apply_env_overrides,
config_snapshot,
@@ -35,7 +36,6 @@ from .builder import (
__all__ = [
"ApiDataSourceConfig",
"BaseDataSourceConfig",
"DbDataSourceConfig",
"JsonlDataSourceConfig",
"PipelineRunConfig",
"DataSourceConfig",
@@ -61,6 +61,7 @@ __all__ = [
"DEFAULT_NODE_TYPES",
"build_default_config",
"build_config",
"build_config_from_file",
"apply_config_overrides",
"apply_env_overrides",
"config_snapshot",
sync_state_machine/config/builder.py
+8
View File
@@ -187,3 +187,11 @@ def build_config(
config = apply_config_overrides(config, overrides)
return config
def build_config_from_file(
*,
project_root: Path,
file_path: Union[str, Path],
) -> PipelineRunConfig:
return build_config(project_root=project_root, overrides_file=Path(file_path))
sync_state_machine/config/datasource_config.py
-6
View File
@@ -22,11 +22,6 @@ class JsonlDataSourceConfig(BaseDataSourceConfig):
read_only: bool = False
class DbDataSourceConfig(BaseDataSourceConfig):
type: Literal["db"] = "db"
class ApiDataSourceConfig(BaseDataSourceConfig):
type: Literal["api"] = "api"
@@ -52,5 +47,4 @@ DataSourceConfig = Annotated[
register_datasource_config_model("jsonl", JsonlDataSourceConfig)
register_datasource_config_model("db", DbDataSourceConfig)
register_datasource_config_model("api", ApiDataSourceConfig)
sync_state_machine/config/pipeline_config.py
-1
View File
@@ -38,7 +38,6 @@ class PipelineRunConfig(BaseModel):
model_config = ConfigDict(extra="forbid", validate_assignment=True)
node_types: List[str] = Field(default_factory=lambda: list(DEFAULT_NODE_TYPES))
target_project_ids: List[str] = Field(default_factory=list)
local_datasource: DataSourceConfig
remote_datasource: DataSourceConfig
sync_state_machine/config/run_presets.py
+2 -39
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
import json
import importlib
from pathlib import Path
from typing import Any, Dict, List, Tuple
@@ -29,43 +28,6 @@ def _replace_placeholders(value: Any, project_root: Path) -> Any:
return value
def _import_from_path(import_path: str) -> Any:
module_path, sep, attr_path = import_path.partition(":")
if not module_path:
raise ValueError(f"Invalid import path: {import_path}")
module = importlib.import_module(module_path)
if not sep:
return module
obj: Any = module
for attr_name in attr_path.split("."):
obj = getattr(obj, attr_name)
return obj
def _apply_datasource_bootstraps(payload: Dict[str, Any]) -> Dict[str, Any]:
extensions = payload.pop("extensions", None)
if extensions is None:
return payload
if not isinstance(extensions, dict):
raise ValueError("extensions must be an object mapping")
bootstrap_paths = extensions.get("datasource_bootstraps") or []
if not isinstance(bootstrap_paths, list):
raise ValueError("extensions.datasource_bootstraps must be a list")
for item in bootstrap_paths:
if not isinstance(item, str) or not item.strip():
raise ValueError("extensions.datasource_bootstraps entries must be non-empty strings")
loaded = _import_from_path(item)
if not callable(loaded):
raise TypeError(f"datasource bootstrap must resolve to a callable: {item}")
loaded()
return payload
def list_preset_names() -> List[str]:
return sorted(PRESET_FILES.keys())
@@ -115,7 +77,8 @@ def load_overrides_from_file(file_path: Path, project_root: Path) -> Dict[str, A
resolved = _replace_placeholders(payload, project_root)
if not isinstance(resolved, dict):
raise ValueError(f"Preset file must contain object mapping: {file_path}")
return _apply_datasource_bootstraps(resolved)
resolved.pop("extensions", None)
return resolved
def load_named_preset_overrides(project_root: Path, preset_name: str) -> Tuple[Dict[str, Any], Path, bool]:
sync_state_machine/config/runtime_config.py
+3 -2
View File
@@ -3,7 +3,7 @@
Prefer importing from `sync_state_machine.config` directly.
"""
from .datasource_config import ApiDataSourceConfig, DbDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
from .datasource_config import ApiDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
from .persist_config import PersistConfig
from .logging_config import LoggingConfig
from .strategy_config import (
@@ -17,6 +17,7 @@ from .pipeline_config import PipelineRunConfig, DEFAULT_NODE_TYPES
from .builder import (
build_default_config,
build_config,
build_config_from_file,
apply_config_overrides,
apply_env_overrides,
config_snapshot,
@@ -24,7 +25,6 @@ from .builder import (
__all__ = [
"ApiDataSourceConfig",
"DbDataSourceConfig",
"JsonlDataSourceConfig",
"DataSourceConfig",
"PersistConfig",
@@ -38,6 +38,7 @@ __all__ = [
"DEFAULT_NODE_TYPES",
"build_default_config",
"build_config",
"build_config_from_file",
"apply_config_overrides",
"apply_env_overrides",
"config_snapshot",
sync_state_machine/datasource/__init__.py
+17 -4
View File
@@ -23,13 +23,25 @@ from .handler import (
from .task_result import TaskResult, TaskStatus
# API DataSource
from .api import ApiClient, ApiDataSource, BaseApiHandler
from .api import ApiClient, ApiDataSource, BaseApiHandler, register as register_api_datasource
# DB DataSource
# DB base helpers
from .db import DbDataSource, BaseDbHandler
# JSONL DataSource
from .jsonl import JsonlDataSource, BaseJsonlHandler
from .jsonl import JsonlDataSource, BaseJsonlHandler, register as register_jsonl_datasource
_BUILTIN_DATASOURCE_TYPES_REGISTERED = False
def ensure_builtin_datasource_types_registered(*, replace: bool = False) -> None:
global _BUILTIN_DATASOURCE_TYPES_REGISTERED
if _BUILTIN_DATASOURCE_TYPES_REGISTERED and not replace:
return
register_jsonl_datasource(replace=replace)
register_api_datasource(replace=replace)
_BUILTIN_DATASOURCE_TYPES_REGISTERED = True
__all__ = [
# 基础类
@@ -46,13 +58,14 @@ __all__ = [
"TaskResult",
"TaskStatus",
"ValidationResult",
"ensure_builtin_datasource_types_registered",
# API DataSource
"ApiClient",
"ApiDataSource",
"BaseApiHandler",
# DB DataSource
# DB base helpers
"DbDataSource",
"BaseDbHandler",
sync_state_machine/datasource/api/__init__.py
+35
View File
@@ -1,15 +1,50 @@
from __future__ import annotations
"""
API DataSource 模块
提供基于 HTTP API 的数据源实现。
"""
from typing import TYPE_CHECKING
from ..type_registry import register_datasource_type
from .client import ApiClient
from .datasource import ApiDataSource
from .handler import BaseApiHandler
if TYPE_CHECKING:
from ...config.datasource_config import ApiDataSourceConfig
def build_api_datasource(config: ApiDataSourceConfig, endpoint_name: str, **_: object):
del endpoint_name
return ApiDataSource(
base_url=config.api_base_url,
uid=config.api_uid,
secret=config.api_secret,
debug=config.api_debug,
poll_max_retries=config.poll_max_retries,
poll_interval=config.poll_interval,
rate_limit_max_requests=config.api_rate_limit_max_requests,
rate_limit_window_seconds=config.api_rate_limit_window_seconds,
)
def register(*, replace: bool = False) -> None:
from ...config.datasource_config import ApiDataSourceConfig
register_datasource_type(
"api",
config_model=ApiDataSourceConfig,
factory=build_api_datasource,
replace=replace,
)
__all__ = [
"ApiClient",
"ApiDataSource",
"BaseApiHandler",
"build_api_datasource",
"register",
]
sync_state_machine/datasource/datasource.py
+12 -1
View File
@@ -88,6 +88,13 @@ 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"
@@ -957,6 +964,10 @@ class BaseDataSource(ABC):
return summary
async def initialize(self) -> None:
"""初始化数据源资源（子类可覆盖）"""
return None
async def close(self) -> None:
"""关闭数据源资源（子类可覆盖以实现特定的清理逻辑）"""
pass # 基类默认空实现
return None
sync_state_machine/datasource/db/__init__.py
+3 -1
View File
@@ -1,4 +1,6 @@
"""DB DataSource 模块。"""
from __future__ import annotations
"""DB DataSource 基类模块。"""
from .datasource import DbDataSource
from .handler import BaseDbHandler
sync_state_machine/datasource/jsonl/__init__.py
+40
View File
@@ -1,13 +1,53 @@
from __future__ import annotations
"""
JSONL DataSource 模块
提供基于 JSONL 文件的数据源实现，用于测试和本地开发。
"""
from pathlib import Path
from typing import TYPE_CHECKING
from ..type_registry import register_datasource_type
from .datasource import JsonlDataSource
from .handler import BaseJsonlHandler
from .path_utils import prepare_remote_jsonl_dir
if TYPE_CHECKING:
from ...config.datasource_config import JsonlDataSourceConfig
def build_jsonl_datasource(
config: JsonlDataSourceConfig,
endpoint_name: str,
*,
project_root: Path | None = None,
):
dir_path = Path(config.jsonl_dir)
if endpoint_name == "remote":
dir_path = prepare_remote_jsonl_dir(dir_path, project_root=project_root or Path.cwd())
elif not dir_path.exists() or not dir_path.is_dir():
raise FileNotFoundError(
f"{endpoint_name}_datasource.jsonl_dir does not exist or is not a directory: {dir_path}"
)
return JsonlDataSource(dir_path, read_only=config.read_only)
def register(*, replace: bool = False) -> None:
from ...config.datasource_config import JsonlDataSourceConfig
register_datasource_type(
"jsonl",
config_model=JsonlDataSourceConfig,
factory=build_jsonl_datasource,
replace=replace,
)
__all__ = [
"JsonlDataSource",
"BaseJsonlHandler",
"build_jsonl_datasource",
"prepare_remote_jsonl_dir",
"register",
]
sync_state_machine/datasource/jsonl/handler.py
+7 -9
View File
@@ -50,24 +50,22 @@ class BaseJsonlHandler(BaseNodeHandler):
):
super().__init__(node_type, node_class, schema)
self.datasource = datasource
self._target_project_ids: Optional[Set[str]] = None
self._target_project_ids: List[str] = []
def set_target_project_ids(self, target_project_ids: List[str]) -> None:
"""注入项目过滤列表，供 load 阶段预过滤使用。"""
if target_project_ids:
self._target_project_ids = {str(pid) for pid in target_project_ids if str(pid)}
else:
self._target_project_ids = 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:
if not self._target_project_ids:
target_project_ids = set(self.datasource.target_project_ids)
if not target_project_ids:
return False
if self.node_type == "project":
return item_id not in self._target_project_ids
return item_id not in target_project_ids
project_id = item.get("project_id")
if project_id:
return str(project_id) not in self._target_project_ids
return str(project_id) not in target_project_ids
return False
async def load(self) -> List['SyncNode']:
sync_state_machine/datasource/jsonl/path_utils.py
+31
View File
@@ -0,0 +1,31 @@
from __future__ import annotations
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
def prepare_remote_jsonl_dir(remote_dir: Path, *, project_root: Path) -> Path:
project_root_path = project_root.resolve()
resolved_remote_dir = remote_dir.resolve(strict=False)
if resolved_remote_dir.exists():
if not resolved_remote_dir.is_dir():
raise FileNotFoundError(
f"remote_datasource.jsonl_dir exists but is not a directory: {resolved_remote_dir}"
)
return resolved_remote_dir
try:
resolved_remote_dir.relative_to(project_root_path)
except ValueError as exc:
raise FileNotFoundError(
"remote_datasource.jsonl_dir does not exist and auto-create is only allowed "
f"under project root: {resolved_remote_dir} (project_root={project_root_path})"
) from exc
resolved_remote_dir.mkdir(parents=True, exist_ok=True)
logger.info("Created missing remote JSONL directory: %s", resolved_remote_dir)
return resolved_remote_dir
sync_state_machine/pipeline/__init__.py
+4 -2
View File
@@ -7,7 +7,6 @@ from .full_sync_pipeline import FullSyncPipeline
from ..config import (
PipelineRunConfig,
ApiDataSourceConfig,
DbDataSourceConfig,
JsonlDataSourceConfig,
DataSourceConfig,
PersistConfig,
@@ -22,7 +21,9 @@ from .factory import (
create_jsonl_to_api_pipeline,
create_api_to_jsonl_pipeline,
create_jsonl_to_jsonl_pipeline,
create_pipeline_from_file,
create_pipeline_from_config,
run_pipeline_from_file,
run_pipeline_from_config,
)
@@ -32,11 +33,12 @@ __all__ = [
"create_jsonl_to_api_pipeline",
"create_jsonl_to_jsonl_pipeline",
"create_api_to_jsonl_pipeline",
"create_pipeline_from_file",
"create_pipeline_from_config",
"run_pipeline_from_file",
"run_pipeline_from_config",
"PipelineRunConfig",
"ApiDataSourceConfig",
"DbDataSourceConfig",
"JsonlDataSourceConfig",
"DataSourceConfig",
"PersistConfig",
sync_state_machine/pipeline/factory.py
+86 -120
View File
@@ -23,19 +23,17 @@ from ..common.binding import BindingManager
from ..common.collection import DataCollection
from ..common.persistence import PersistenceBackend
from ..common.registry import DomainRegistry
from ..datasource.api.datasource import ApiDataSource
from ..datasource.db.datasource import DbDataSource
from ..datasource.jsonl.datasource import JsonlDataSource
from ..datasource.type_registry import get_datasource_factory, register_datasource_factory
from ..datasource import ensure_builtin_datasource_types_registered
from ..datasource.type_registry import get_datasource_factory
from ..config import (
PipelineRunConfig,
DataSourceConfig,
ApiDataSourceConfig,
DbDataSourceConfig,
JsonlDataSourceConfig,
PersistConfig,
JsonlDataSourceConfig,
LoggingConfig,
StrategyRuntimeConfig,
build_config_from_file,
apply_config_overrides,
OrphanAction,
UpdateDirection,
@@ -55,27 +53,9 @@ def _project_root() -> Path:
def _prepare_remote_jsonl_dir(remote_dir: Path, *, project_root: Optional[Path] = None) -> Path:
project_root_path = (project_root or _project_root()).resolve()
resolved_remote_dir = remote_dir.resolve(strict=False)
from ..datasource.jsonl.path_utils import prepare_remote_jsonl_dir
if resolved_remote_dir.exists():
if not resolved_remote_dir.is_dir():
raise FileNotFoundError(
f"remote_datasource.jsonl_dir exists but is not a directory: {resolved_remote_dir}"
)
return resolved_remote_dir
try:
resolved_remote_dir.relative_to(project_root_path)
except ValueError as exc:
raise FileNotFoundError(
"remote_datasource.jsonl_dir does not exist and auto-create is only allowed "
f"under project root: {resolved_remote_dir} (project_root={project_root_path})"
) from exc
resolved_remote_dir.mkdir(parents=True, exist_ok=True)
logger.info("Created missing remote JSONL directory: %s", resolved_remote_dir)
return resolved_remote_dir
return prepare_remote_jsonl_dir(remote_dir, project_root=(project_root or _project_root()))
def _ensure_pipeline_logger(
@@ -104,13 +84,8 @@ def _ensure_pipeline_logger(
def _build_handler_config(
node_type: str,
ds_config: DataSourceConfig,
target_project_ids: List[str],
) -> Dict[str, Any]:
config = dict(ds_config.handler_configs.get(node_type, {}))
if node_type == "project" and target_project_ids:
config.setdefault("filter_project_id", target_project_ids)
return config
return dict(ds_config.handler_configs.get(node_type, {}))
def _create_handler(
@@ -118,7 +93,6 @@ def _create_handler(
node_type: str,
ds_config: DataSourceConfig,
datasource,
target_project_ids: List[str],
):
datasource_type = ds_config.type
handler_class = DomainRegistry.get_handler(node_type, datasource_type)
@@ -127,7 +101,7 @@ def _create_handler(
f"Missing {datasource_type.upper()} handler for node_type: {node_type}"
)
handler_config = _build_handler_config(node_type, ds_config, target_project_ids)
handler_config = _build_handler_config(node_type, ds_config)
if datasource_type == "api":
handler = handler_class(**handler_config)
else:
@@ -138,22 +112,12 @@ def _create_handler(
return handler
def _resolve_datasource_target_project_ids(config: PipelineRunConfig) -> tuple[List[str], List[str]]:
fallback = [str(pid) for pid in config.target_project_ids if str(pid)]
local_cfg_ids = [str(pid) for pid in config.local_datasource.target_project_ids if str(pid)]
remote_cfg_ids = [str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)]
local_target_ids = local_cfg_ids if local_cfg_ids else fallback
remote_target_ids = remote_cfg_ids if remote_cfg_ids else fallback
return local_target_ids, remote_target_ids
async def _create_datasource(
ds_config: DataSourceConfig,
*,
datasource_override=None,
endpoint_name: str,
project_root: Path | None = None,
):
if datasource_override is not None:
return datasource_override
@@ -162,54 +126,24 @@ async def _create_datasource(
if datasource_factory is None:
raise RuntimeError(f"No datasource factory registered for type: {ds_config.type}")
datasource = datasource_factory(ds_config, endpoint_name)
factory_kwargs: dict[str, Any] = {}
signature = inspect.signature(datasource_factory)
if "project_root" in signature.parameters:
factory_kwargs["project_root"] = project_root or _project_root()
datasource = datasource_factory(ds_config, endpoint_name, **factory_kwargs)
if inspect.isawaitable(datasource):
datasource = await datasource
return datasource
def _build_jsonl_datasource(ds_config: JsonlDataSourceConfig, endpoint_name: str):
dir_path = Path(ds_config.jsonl_dir)
if endpoint_name == "remote":
dir_path = _prepare_remote_jsonl_dir(dir_path)
elif not dir_path.exists() or not dir_path.is_dir():
raise FileNotFoundError(
f"{endpoint_name}_datasource.jsonl_dir does not exist or is not a directory: {dir_path}"
)
return JsonlDataSource(dir_path, read_only=ds_config.read_only)
async def _initialize_datasource(datasource, *, endpoint_name: str) -> None:
try:
await datasource.initialize()
except Exception as exc:
raise RuntimeError(f"Failed to initialize {endpoint_name} datasource: {type(exc).__name__}: {exc}") from exc
def _build_db_datasource(ds_config: DbDataSourceConfig, endpoint_name: str):
del ds_config, endpoint_name
return DbDataSource()
async def _build_api_datasource(ds_config: ApiDataSourceConfig, endpoint_name: str):
del endpoint_name
datasource = ApiDataSource(
base_url=ds_config.api_base_url,
uid=ds_config.api_uid,
secret=ds_config.api_secret,
debug=ds_config.api_debug,
poll_max_retries=ds_config.poll_max_retries,
poll_interval=ds_config.poll_interval,
rate_limit_max_requests=ds_config.api_rate_limit_max_requests,
rate_limit_window_seconds=ds_config.api_rate_limit_window_seconds,
)
await datasource.initialize()
return datasource
def _ensure_builtin_datasource_factories_registered() -> None:
if get_datasource_factory("jsonl") is None:
register_datasource_factory("jsonl", _build_jsonl_datasource)
if get_datasource_factory("db") is None:
register_datasource_factory("db", _build_db_datasource)
if get_datasource_factory("api") is None:
register_datasource_factory("api", _build_api_datasource)
_ensure_builtin_datasource_factories_registered()
ensure_builtin_datasource_types_registered()
def _register_handlers_for_endpoint(
@@ -217,7 +151,6 @@ def _register_handlers_for_endpoint(
datasource,
ds_config: DataSourceConfig,
node_types: List[str],
target_project_ids: List[str],
) -> None:
for node_type in node_types:
datasource.register_handler(
@@ -225,7 +158,6 @@ def _register_handlers_for_endpoint(
node_type=node_type,
ds_config=ds_config,
datasource=datasource,
target_project_ids=target_project_ids,
)
)
@@ -240,13 +172,11 @@ def _register_handlers(
datasource=local_ds,
ds_config=config.local_datasource,
node_types=all_types,
target_project_ids=[str(pid) for pid in config.local_datasource.target_project_ids if str(pid)],
)
_register_handlers_for_endpoint(
datasource=remote_ds,
ds_config=config.remote_datasource,
node_types=all_types,
target_project_ids=[str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)],
)
@@ -318,7 +248,6 @@ async def create_pipeline_from_config(
all_types = list(config.node_types)
strategy_map = _resolve_strategy_map(config, all_types)
local_target_project_ids, remote_target_project_ids = _resolve_datasource_target_project_ids(config)
if config.persist.backend != "sqlite":
raise NotImplementedError(f"Persistence backend not supported yet: {config.persist.backend}")
@@ -338,31 +267,20 @@ async def create_pipeline_from_config(
config.local_datasource,
datasource_override=local_datasource_override,
endpoint_name="local",
project_root=_project_root(),
)
await _initialize_datasource(local_ds, endpoint_name="local")
remote_ds = await _create_datasource(
config.remote_datasource,
datasource_override=remote_datasource_override,
endpoint_name="remote",
project_root=_project_root(),
)
local_cfg_with_target = config.local_datasource.model_copy(
update={"target_project_ids": list(local_target_project_ids)},
deep=True,
)
remote_cfg_with_target = config.remote_datasource.model_copy(
update={"target_project_ids": list(remote_target_project_ids)},
deep=True,
)
handler_target_config = config.model_copy(
update={
"local_datasource": local_cfg_with_target,
"remote_datasource": remote_cfg_with_target,
},
deep=True,
)
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(handler_target_config, local_ds, remote_ds, all_types)
local_ds.set_target_project_ids(local_target_project_ids)
remote_ds.set_target_project_ids(remote_target_project_ids)
_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)
@@ -387,9 +305,6 @@ async def create_pipeline_from_config(
remote_collection=remote_collection,
binding_manager=binding_manager,
enable_persistence=config.persist.enable,
target_project_ids=config.target_project_ids,
local_target_project_ids=local_target_project_ids,
remote_target_project_ids=remote_target_project_ids,
)
return pipeline
@@ -474,6 +389,40 @@ async def run_pipeline_from_config(
await pipeline.close()
async def create_pipeline_from_file(
*,
project_root: Path,
config_path: str | Path,
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,
local_datasource_override=local_datasource_override,
remote_datasource_override=remote_datasource_override,
)
async def run_pipeline_from_file(
*,
project_root: Path,
config_path: str | Path,
title: str = "sync_state_machine pipeline",
print_summary: bool = True,
local_datasource_override=None,
remote_datasource_override=None,
) -> Dict[str, Any]:
config = build_config_from_file(project_root=project_root, file_path=config_path)
return await run_pipeline_from_config(
config,
title=title,
print_summary=print_summary,
local_datasource_override=local_datasource_override,
remote_datasource_override=remote_datasource_override,
)
async def create_jsonl_to_api_pipeline(
*,
# 路径配置
@@ -506,8 +455,12 @@ async def create_jsonl_to_api_pipeline(
) -> FullSyncPipeline:
cfg = PipelineRunConfig(
node_types=node_types,
target_project_ids=target_project_ids or [],
local_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=local_jsonl_dir, read_only=False),
local_datasource=JsonlDataSourceConfig(
type="jsonl",
jsonl_dir=local_jsonl_dir,
read_only=False,
target_project_ids=target_project_ids or [],
),
remote_datasource=ApiDataSourceConfig(
type="api",
api_base_url=api_base_url,
@@ -570,7 +523,6 @@ async def create_api_to_jsonl_pipeline(
) -> FullSyncPipeline:
cfg = PipelineRunConfig(
node_types=node_types,
target_project_ids=target_project_ids or [],
local_datasource=ApiDataSourceConfig(
type="api",
api_base_url=api_base_url,
@@ -584,7 +536,12 @@ async def create_api_to_jsonl_pipeline(
target_project_ids=target_project_ids or [],
handler_configs=handler_configs or {},
),
remote_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=remote_jsonl_dir, read_only=False),
remote_datasource=JsonlDataSourceConfig(
type="jsonl",
jsonl_dir=remote_jsonl_dir,
read_only=False,
target_project_ids=target_project_ids or [],
),
strategies=[],
persist=PersistConfig(
db_path=persistence_db,
@@ -618,9 +575,18 @@ async def create_jsonl_to_jsonl_pipeline(
) -> FullSyncPipeline:
cfg = PipelineRunConfig(
node_types=node_types,
target_project_ids=target_project_ids or [],
local_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=local_jsonl_dir, read_only=False),
remote_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=remote_jsonl_dir, read_only=False),
local_datasource=JsonlDataSourceConfig(
type="jsonl",
jsonl_dir=local_jsonl_dir,
read_only=False,
target_project_ids=target_project_ids or [],
),
remote_datasource=JsonlDataSourceConfig(
type="jsonl",
jsonl_dir=remote_jsonl_dir,
read_only=False,
target_project_ids=target_project_ids or [],
),
strategies=[],
persist=PersistConfig(
db_path=persistence_db,
sync_state_machine/pipeline/full_sync_pipeline.py
-21
View File
@@ -51,10 +51,6 @@ class FullSyncPipeline:
load_order: Optional[List[str]] = None,
sync_order: Optional[List[str]] = None,
enable_persistence: bool = True,
# 项目过滤
target_project_ids: Optional[List[str]] = None,
local_target_project_ids: Optional[List[str]] = None,
remote_target_project_ids: Optional[List[str]] = None,
):
# 验证必需参数（由工厂函数创建）
if strategies is None:
@@ -70,17 +66,6 @@ class FullSyncPipeline:
self.sync_order = sync_order or self.node_types
self.persistence = persistence
self.enable_persistence = enable_persistence
self.target_project_ids = target_project_ids or []
self.local_target_project_ids = (
list(local_target_project_ids)
if local_target_project_ids is not None
else list(self.target_project_ids)
)
self.remote_target_project_ids = (
list(remote_target_project_ids)
if remote_target_project_ids is not None
else list(self.target_project_ids)
)
self.local_collection = local_collection
self.remote_collection = remote_collection
@@ -187,12 +172,6 @@ class FullSyncPipeline:
await self._load_from_datasource()
self._logger.info("✅ Bootstrap: datasource fetched")
if self.local_target_project_ids or self.remote_target_project_ids:
local_deleted = await self.local_collection.filter_by_project_ids(self.local_target_project_ids)
remote_deleted = await self.remote_collection.filter_by_project_ids(self.remote_target_project_ids)
if local_deleted > 0 or remote_deleted > 0:
self._logger.info(f"🔍 Project filtering: deleted {local_deleted} local, {remote_deleted} remote nodes")
async def phase_bind(self) -> None:
strategy_map = {s.node_type: s for s in self.strategies}
for node_type in self.sync_order: