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 17:09:46 +08:00
parent 6ae54e3edd
commit 32fa374e86
20 changed files with 63 additions and 979 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/legacy/run_full_sync_pipeline_jsonl.py
-208
View File
@@ -1,208 +0,0 @@
"""
Run FullSyncPipeline: JSONL -> JSONL (filtered_datasource -> remote_datasource)
直接运行脚本，不使用 pytest。
"""
from __future__ import annotations
import asyncio
import sys
from pathlib import Path
from typing import List
# 确保能导入项目模块
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
# 触发 DomainRegistry 注册
import sync_system_new.domain # noqa: F401, E402
from sync_system_new.common.binding import BindingManager # noqa: E402
from sync_system_new.common.collection import DataCollection # noqa: E402
from sync_system_new.common.persistence import PersistenceBackend # noqa: E402
from sync_system_new.common.registry import DomainRegistry # noqa: E402
from sync_system_new.datasource.jsonl.datasource import JsonlDataSource # noqa: E402
from sync_system_new.pipeline import FullSyncPipeline # noqa: E402
from sync_system_new.sync_system.config import OrphanAction, UpdateDirection # noqa: E402
LOCAL_DIR = PROJECT_ROOT / "filtered_datasource" / "datasource" / "ecm-2025-12-02"
REMOTE_DIR = PROJECT_ROOT / "remote_datasource" / "datasource"
PERSISTENCE_DB = PROJECT_ROOT / "test_results" / "full_sync_pipeline_jsonl_run.db"
WIPE_DB_ON_START = False
NODE_TYPES: List[str] = [
"company",
# "supplier",
"user",
"project",
"project_detail_data",
"contract",
"contract_change",
"construction_task",
"construction_task_detail",
"material",
"material_detail",
"contract_settlement",
"contract_settlement_detail",
"personnel",
"personnel_detail",
"preparation",
"production",
"lar",
"units",
]
class StrategyLogger:
"""包装策略，打印执行完成信息。"""
def __init__(self, inner):
self.inner = inner
self.node_type = inner.node_type
async def bind(self):
return await self.inner.bind()
async def create(self):
created = await self.inner.create()
print(f"✅ Strategy create: {self.node_type} (created={len(created)})")
return created
async def update(self):
updated = await self.inner.update()
print(f"✅ Strategy update: {self.node_type} (updated={len(updated)})")
return updated
async def main() -> None:
PERSISTENCE_DB.parent.mkdir(parents=True, exist_ok=True)
if WIPE_DB_ON_START and PERSISTENCE_DB.exists():
PERSISTENCE_DB.unlink()
persistence = PersistenceBackend(str(PERSISTENCE_DB))
await persistence.initialize()
local_ds = JsonlDataSource(LOCAL_DIR, read_only=True)
remote_ds = JsonlDataSource(REMOTE_DIR, read_only=False)
for node_type in NODE_TYPES:
handler_class = DomainRegistry.get_jsonl_handler(node_type)
if handler_class is None:
raise RuntimeError(f"Missing JSONL handler for node_type: {node_type}")
local_ds.register_handler(handler_class(local_ds))
remote_ds.register_handler(handler_class(remote_ds))
local_collection = DataCollection("local", persistence, auto_persist=False)
remote_collection = DataCollection("remote", persistence, auto_persist=False)
binding_manager = BindingManager(persistence, auto_persist=False)
strategies = []
for node_type in NODE_TYPES:
strategy_class = DomainRegistry.get_strategy_class(node_type)
if strategy_class is None:
raise RuntimeError(f"Missing strategy for node_type: {node_type}")
strategy = strategy_class(node_type, local_collection, remote_collection, binding_manager)
# 运行时强制只推不拉（不覆盖原有 depend_fields/auto_bind_fields）
strategy.update_config(
local_orphan_action=OrphanAction.CREATE_REMOTE,
remote_orphan_action=OrphanAction.NONE,
update_direction=UpdateDirection.PUSH,
)
if node_type == "contract":
# contract 仅依赖 project_id，避免 company_id 为空导致依赖阻断
strategy.update_config(depend_fields={"project_id": "project"})
strategies.append(StrategyLogger(strategy))
pipeline = FullSyncPipeline(
local_datasource=local_ds,
remote_datasource=remote_ds,
strategies=strategies,
node_types=NODE_TYPES,
load_order=NODE_TYPES,
sync_order=NODE_TYPES,
persistence=persistence,
local_collection=local_collection,
remote_collection=remote_collection,
binding_manager=binding_manager,
enable_persistence=True,
fail_fast_on_precheck=False,
)
stats = await pipeline.run()
print(f"Pipeline finished. Stats: {stats}")
binding_counts = {}
for node_type in NODE_TYPES:
records = await binding_manager.get_all_records(node_type)
binding_counts[node_type] = len(records)
def format_action(summary, key: str) -> str:
action = summary.get(key, {})
return f"{action.get('total', 0)}({action.get('success', 0)}/{action.get('failed', 0)})"
def format_total(summary) -> str:
total = summary.get("total", {})
return f"{total.get('total', 0)}({total.get('success', 0)}/{total.get('failed', 0)})"
def print_summary(title: str, action_summary: dict) -> None:
print(f"\n=== {title} summary ===")
print("doc_type: bindings | create(s/f) | update(s/f) | delete(s/f) | none | total(s/f)")
total_bindings = 0
total_create = {"total": 0, "success": 0, "failed": 0}
total_update = {"total": 0, "success": 0, "failed": 0}
total_delete = {"total": 0, "success": 0, "failed": 0}
total_none = 0
total_all = {"total": 0, "success": 0, "failed": 0}
for node_type in NODE_TYPES:
summary = action_summary.get(node_type, {})
bindings = binding_counts.get(node_type, 0)
create_str = format_action(summary, "create")
update_str = format_action(summary, "update")
delete_str = format_action(summary, "delete")
none_count = summary.get("none", {}).get("total", 0)
total_str = format_total(summary)
print(
f"{node_type}: bindings={bindings} "
f"create={create_str} update={update_str} delete={delete_str} "
f"none={none_count} total={total_str}"
)
total_bindings += bindings
for bucket, key in (
(total_create, "create"),
(total_update, "update"),
(total_delete, "delete"),
):
action = summary.get(key, {})
bucket["total"] += action.get("total", 0)
bucket["success"] += action.get("success", 0)
bucket["failed"] += action.get("failed", 0)
total_none += none_count
total_all["total"] += summary.get("total", {}).get("total", 0)
total_all["success"] += summary.get("total", {}).get("success", 0)
total_all["failed"] += summary.get("total", {}).get("failed", 0)
print(
"TOTAL: "
f"bindings={total_bindings} "
f"create={total_create['total']}({total_create['success']}/{total_create['failed']}) "
f"update={total_update['total']}({total_update['success']}/{total_update['failed']}) "
f"delete={total_delete['total']}({total_delete['success']}/{total_delete['failed']}) "
f"none={total_none} "
f"total={total_all['total']}({total_all['success']}/{total_all['failed']})"
)
print_summary("Local", local_ds.get_action_summary())
print_summary("Remote", remote_ds.get_action_summary())
await persistence.close()
if __name__ == "__main__":
asyncio.run(main())
scripts/legacy/run_full_sync_pipeline_jsonl_sm.py
-153
View File
@@ -1,153 +0,0 @@
"""
run_full_sync_pipeline_jsonl_sm.py
独立执行：基于 sync_state_machine 的 JSONL -> JSONL 全量同步脚本。
用途：纯本地可复现，不依赖远端 API；用于验证重构后 pipeline 主流程可跑通。
"""
from __future__ import annotations
import asyncio
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List
from pydantic import BaseModel, ConfigDict, Field
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import sync_state_machine.domain # noqa: F401, E402
from sync_state_machine.pipeline import create_jsonl_to_jsonl_pipeline # noqa: E402
class JsonlSmConfig(BaseModel):
model_config = ConfigDict(extra="forbid", validate_assignment=True)
local_jsonl_dir: str = str(PROJECT_ROOT / "filtered_datasource" / "datasource" / "ecm-2025-12-02")
remote_jsonl_dir: str = str(PROJECT_ROOT / "remote_datasource" / "datasource")
persistence_db: str = str(PROJECT_ROOT / "test_results" / "sync_state_machine" / "full_sync_pipeline_jsonl_sm.db")
wipe_persistence_on_start: bool = False
target_project_ids: List[str] = Field(default_factory=lambda: ["f3e02e84-e81c-4aa6-88d4-f5aa108c8227"])
project_auto_bind_fields: List[str] = Field(default_factory=lambda: ["name"])
project_id_bindings: Dict[str, str] = Field(default_factory=dict)
node_types: List[str] = Field(default_factory=lambda: [
"company",
"user",
"project",
"contract",
"contract_change",
"construction_task",
"construction_task_detail",
"material",
"material_detail",
"contract_settlement",
"contract_settlement_detail",
"personnel",
"personnel_detail",
"preparation",
"production",
"lar",
"units",
])
def _load_dotenv() -> Dict[str, str]:
dotenv_path = PROJECT_ROOT / ".env"
if not dotenv_path.exists():
return {}
values: Dict[str, str] = {}
for raw_line in dotenv_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip().strip('"').strip("'")
return values
def build_config() -> JsonlSmConfig:
cfg = JsonlSmConfig()
env = _load_dotenv()
cfg.wipe_persistence_on_start = (
os.getenv("SPS_JSONL_WIPE_DB", env.get("SPS_JSONL_WIPE_DB", str(cfg.wipe_persistence_on_start))).lower()
in {"1", "true", "yes"}
)
return cfg
def _print_config_snapshot(config: JsonlSmConfig) -> None:
snapshot = {
"local_jsonl_dir": config.local_jsonl_dir,
"remote_jsonl_dir": config.remote_jsonl_dir,
"persistence_db": config.persistence_db,
"wipe_persistence_on_start": config.wipe_persistence_on_start,
"target_project_ids": config.target_project_ids,
"project_auto_bind_fields": config.project_auto_bind_fields,
"project_id_bindings": config.project_id_bindings,
"node_types": config.node_types,
}
print(f"🔧 Resolved Config: {snapshot}")
async def main() -> int:
pipeline = None
log_path = None
config = build_config()
try:
log_dir = PROJECT_ROOT / "test_results" / "sync_state_machine"
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / f"jsonl_pipeline_sm_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
print("\n" + "=" * 80)
print("🚀 Creating sync_state_machine JSONL -> JSONL pipeline")
print(f"📝 Log file: {log_path}")
_print_config_snapshot(config)
print("=" * 80)
Path(config.persistence_db).parent.mkdir(parents=True, exist_ok=True)
pipeline = await create_jsonl_to_jsonl_pipeline(
local_jsonl_dir=config.local_jsonl_dir,
remote_jsonl_dir=config.remote_jsonl_dir,
persistence_db=config.persistence_db,
node_types=config.node_types,
load_order=config.node_types,
sync_order=config.node_types,
enable_persistence=True,
wipe_persistence_on_start=config.wipe_persistence_on_start,
target_project_ids=config.target_project_ids,
project_id_bindings=config.project_id_bindings,
project_auto_bind_fields=config.project_auto_bind_fields,
enable_project_bind_override=True,
enable_project_filter_override=True,
initialize_logger=True,
logger_file_path=str(log_path),
)
stats = await pipeline.run()
print(f"\n✅ Pipeline completed: stats={stats}")
print(f"🗃️ Persistence DB: {config.persistence_db}")
return 0
except KeyboardInterrupt:
print("\n⚠️ Interrupted by user")
return 130
except Exception as exc:
print(f"\n❌ Pipeline failed: {type(exc).__name__}: {exc}")
import traceback
traceback.print_exc()
return 1
finally:
if pipeline is not None:
await pipeline.close()
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))
scripts/legacy/run_full_sync_pipeline_simple.py
-179
View File
@@ -1,179 +0,0 @@
"""
run_full_sync_pipeline_simple.py
极简测试脚本：演示使用工厂函数创建 Pipeline
- 只需配置参数
- 调用工厂函数
- 执行 pipeline.run()
"""
import asyncio
import sys
from pathlib import Path
# 项目路径
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
# 触发 Domain 注册
import sync_system_new.domain # noqa: F401, E402
from sync_system_new.pipeline import create_jsonl_to_api_pipeline # noqa: E402
# ========== 配置参数 ==========
CONFIG = {
# 路径
"local_jsonl_dir": str(PROJECT_ROOT / "filtered_datasource" / "datasource" / "ecm-2025-12-02"),
"persistence_db": str(PROJECT_ROOT / "_runtime" / "test_simple.db"),
# API
"api_base_url": "http://localhost:8000/api/third/v2",
"api_uid": "6b18c681-92d5-4c7f-802c-df413b6504c8",
"api_secret": "1a28Az5dBD1Gc3URpmHwiR0XfzW7UtScQnqJQzgpzdg",
# 节点类型
"node_types": [
"company",
"supplier",
# "user",
"project",
"project_detail_data",
"preparation",
"production",
"lar",
"lar_change",
"units",
"contract",
"contract_change",
"construction_task",
"construction_task_detail",
"material",
"material_detail",
"contract_settlement",
"contract_settlement_detail",
"personnel",
"personnel_detail",
],
# 项目过滤和预绑定
"target_project_ids": ["f3e02e84-e81c-4aa6-88d4-f5aa108c8227"],
"project_id_bindings": {
"f3e02e84-e81c-4aa6-88d4-f5aa108c8227": "f3e02e84-e81c-4aa6-88d4-f5aa108c8227",
},
# Handler 自定义配置
"handler_configs": {
# supplier: 从持久化加载，不调用 API（后续运行时启用）
"supplier": {"load_mode": "persisted_only"},
},
# Skip/Force Sync 控制
# skip_sync_types: 跳过同步（只执行 bind，不执行 create/update）
# 后续运行时可配置: "skip_sync_types": ["supplier"]
"skip_sync_types": ["supplier"],
# 行为
"wipe_persistence_on_start": False, # 改为 False，否则持久化数据会被清空
"enable_persistence": True,
# API 调试
"api_debug": True,
"poll_max_retries": 1,
"poll_interval": 2.0,
}
async def main():
"""主函数 - 返回 True 表示成功，False 表示被中断"""
from datetime import datetime
import sys
from pathlib import Path
pipeline = None
interrupted = False
log_file = None
original_stdout = None
original_stderr = None
# 日志设置
log_dir = Path(PROJECT_ROOT) / "test_results"
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / f"simple_pipeline_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
try:
# 重定向输出到日志文件
log_file = open(log_path, "w", encoding="utf-8")
class Tee:
def __init__(self, *streams):
self._streams = streams
def write(self, data):
for stream in self._streams:
stream.write(data)
stream.flush()
return len(data)
def flush(self):
for stream in self._streams:
stream.flush()
original_stdout = sys.stdout
original_stderr = sys.stderr
sys.stdout = Tee(original_stdout, log_file)
sys.stderr = Tee(original_stderr, log_file)
print("\n" + "=" * 60)
print("🚀 Creating Pipeline...")
print(f"📝 Log file: {log_path}")
print("=" * 60)
# 一行创建完整 Pipeline
pipeline = await create_jsonl_to_api_pipeline(**CONFIG)
print(f"✅ Pipeline created!")
print(f" - Node types: {len(pipeline.node_types)}")
print(f" - Strategies: {len(pipeline.strategies)}")
print("\n" + "=" * 60)
print("🏃 Running Pipeline...")
print("=" * 60)
# 执行同步（print_summary 已整合到 stage4_post_check 中）
stats = await pipeline.run()
print("\n✅ Done!")
except KeyboardInterrupt:
print("\n\n⚠️ KeyboardInterrupt received, shutting down gracefully...")
interrupted = True
except Exception as e:
print(f"\n\n❌ Error: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
finally:
if pipeline is not None:
print("🧹 Cleaning up resources...")
await pipeline.close()
print("✅ Resources cleaned up")
# 恢复原始输出
if original_stdout is not None:
sys.stdout = original_stdout
if original_stderr is not None:
sys.stderr = original_stderr
if log_file is not None:
log_file.close()
if not interrupted:
print(f"\n📝 Log saved to: {log_path}")
return not interrupted # 返回 False 表示被中断
if __name__ == "__main__":
try:
success = asyncio.run(main())
sys.exit(0 if success else 130) # 130 是标准的 Ctrl+C 退出码
except KeyboardInterrupt:
# asyncio.run() 可能会重新抛出 KeyboardInterrupt
print("\n⚠️ Program interrupted")
sys.exit(130)
scripts/legacy/run_full_sync_pipeline_simple_sm.py
-143
View File
@@ -1,143 +0,0 @@
"""
run_full_sync_pipeline_simple_sm.py
独立执行：基于 sync_state_machine 的 JSONL -> API Pipeline。
用途：保留线上接口参数，直接执行重构后的推送链路（非 pytest）。
"""
from __future__ import annotations
import asyncio
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List
from pydantic import BaseModel, ConfigDict, Field
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import sync_state_machine.domain # noqa: F401, E402
from sync_state_machine.pipeline import create_jsonl_to_api_pipeline # noqa: E402
class SimpleSmConfig(BaseModel):
model_config = ConfigDict(extra="forbid", validate_assignment=True)
local_jsonl_dir: str = str(PROJECT_ROOT / "filtered_datasource" / "datasource" / "ecm-2025-12-02")
persistence_db: str = str(PROJECT_ROOT / "test_results" / "sync_state_machine" / "simple_pipeline_sm.db")
api_base_url: str = "http://localhost:8000/api/third/v2"
api_uid: str = "6b18c681-92d5-4c7f-802c-df413b6504c8"
api_secret: str = "1a28Az5dBD1Gc3URpmHwiR0XfzW7UtScQnqJQzgpzdg"
node_types: List[str] = Field(default_factory=lambda: [
"project",
"contract",
"contract_settlement",
])
target_project_ids: List[str] = Field(default_factory=lambda: ["f3e02e84-e81c-4aa6-88d4-f5aa108c8227"])
project_id_bindings: Dict[str, str] = Field(
default_factory=lambda: {
"f3e02e84-e81c-4aa6-88d4-f5aa108c8227": "f3e02e84-e81c-4aa6-88d4-f5aa108c8227"
}
)
skip_sync_types: List[str] = Field(default_factory=lambda: ["supplier"])
wipe_persistence_on_start: bool = False
enable_persistence: bool = True
api_debug: bool = True
poll_max_retries: int = 1
poll_interval: float = 2.0
def _load_dotenv() -> Dict[str, str]:
dotenv_path = PROJECT_ROOT / ".env"
if not dotenv_path.exists():
return {}
values: Dict[str, str] = {}
for raw_line in dotenv_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip().strip('"').strip("'")
return values
def build_config() -> SimpleSmConfig:
cfg = SimpleSmConfig()
env = _load_dotenv()
cfg.api_base_url = os.getenv("SPS_API_BASE_URL", env.get("SPS_API_BASE_URL", cfg.api_base_url))
cfg.api_uid = os.getenv("SPS_API_UID", env.get("SPS_API_UID", cfg.api_uid))
cfg.api_secret = os.getenv("SPS_API_SECRET", env.get("SPS_API_SECRET", cfg.api_secret))
return cfg
def _print_config_snapshot(config: SimpleSmConfig) -> None:
masked_secret = "***" if config.api_secret else ""
snapshot = {
"local_jsonl_dir": config.local_jsonl_dir,
"persistence_db": config.persistence_db,
"api_base_url": config.api_base_url,
"api_uid": config.api_uid,
"api_secret": masked_secret,
"node_types": config.node_types,
"target_project_ids": config.target_project_ids,
"project_id_bindings": config.project_id_bindings,
"skip_sync_types": config.skip_sync_types,
"wipe_persistence_on_start": config.wipe_persistence_on_start,
"enable_persistence": config.enable_persistence,
"api_debug": config.api_debug,
"poll_max_retries": config.poll_max_retries,
"poll_interval": config.poll_interval,
}
print(f"🔧 Resolved Config: {snapshot}")
async def main() -> int:
pipeline = None
log_path = None
config = build_config()
try:
log_dir = PROJECT_ROOT / "test_results" / "sync_state_machine"
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / f"simple_pipeline_sm_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
print("\n" + "=" * 80)
print("🚀 Creating sync_state_machine JSONL -> API pipeline")
print(f"📝 Log file: {log_path}")
_print_config_snapshot(config)
print("=" * 80)
pipeline = await create_jsonl_to_api_pipeline(
**config.model_dump(),
initialize_logger=True,
logger_file_path=str(log_path),
)
stats = await pipeline.run()
print(f"\n✅ Pipeline completed: stats={stats}")
print(f"🗃️ Persistence DB: {config.persistence_db}")
return 0
except KeyboardInterrupt:
print("\n⚠️ Interrupted by user")
return 130
except Exception as exc:
print(f"\n❌ Pipeline failed: {type(exc).__name__}: {exc}")
import traceback
traceback.print_exc()
return 1
finally:
if pipeline is not None:
await pipeline.close()
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))
sync_state_machine/__init__.py
+1 -3
View File
@@ -29,7 +29,7 @@ from .config import (
register_datasource_factory,
register_datasource_type,
)
from .datasource import ApiClient, ApiDataSource, BaseApiHandler, BaseDbHandler, BaseJsonlHandler, DbDataSource, JsonlDataSource
from .datasource import ApiClient, ApiDataSource, BaseApiHandler, BaseJsonlHandler, JsonlDataSource
from .pipeline import FullSyncPipeline, create_pipeline_from_config, run_pipeline_from_config
try:
@@ -64,10 +64,8 @@ __all__ = [
"register_datasource_type",
"ApiClient",
"ApiDataSource",
"DbDataSource",
"JsonlDataSource",
"BaseApiHandler",
"BaseDbHandler",
"BaseJsonlHandler",
"FullSyncPipeline",
"create_pipeline_from_config",
sync_state_machine/config/__init__.py
-2
View File
@@ -29,7 +29,6 @@ from .builder import (
build_config,
build_config_from_file,
apply_config_overrides,
apply_env_overrides,
config_snapshot,
)
@@ -63,6 +62,5 @@ __all__ = [
"build_config",
"build_config_from_file",
"apply_config_overrides",
"apply_env_overrides",
"config_snapshot",
]
sync_state_machine/config/builder.py
+1 -14
View File
@@ -107,7 +107,6 @@ def build_default_config(project_root: Path) -> PipelineRunConfig:
"read_only": False,
"handler_configs": {},
},
"target_project_ids": [],
"persist": {
"db_path": str(project_root / "test_results" / "sync_state_machine" / "pipeline_default.db"),
"enable": True,
@@ -129,29 +128,17 @@ def build_default_config(project_root: Path) -> PipelineRunConfig:
def apply_config_overrides(config: PipelineRunConfig, overrides: Dict[str, Any]) -> PipelineRunConfig:
merged = _deep_merge(config.model_dump(), overrides)
raw_strategies = merged.pop("strategies", None)
legacy_strategy_overrides = merged.pop("strategy_overrides", None)
resolved = PipelineRunConfig.model_validate(merged)
if raw_strategies is not None:
resolved.strategies = _apply_strategy_overrides(resolved.node_types, raw_strategies)
else:
resolved.strategies = _apply_strategy_overrides(resolved.node_types, legacy_strategy_overrides or {})
resolved.strategies = _apply_strategy_overrides(resolved.node_types, {})
return resolved
def apply_env_overrides(
config: PipelineRunConfig,
project_root: Path,
*,
dotenv_file: Optional[Union[str, Path]] = None,
) -> PipelineRunConfig:
raise NotImplementedError(
"Environment-based overrides are removed. Use config files via overrides_file instead."
)
def build_config(
*,
project_root: Path,
sync_state_machine/config/runtime_config.py
-2
View File
@@ -19,7 +19,6 @@ from .builder import (
build_config,
build_config_from_file,
apply_config_overrides,
apply_env_overrides,
config_snapshot,
)
@@ -40,6 +39,5 @@ __all__ = [
"build_config",
"build_config_from_file",
"apply_config_overrides",
"apply_env_overrides",
"config_snapshot",
]
sync_state_machine/datasource/__init__.py
-7
View File
@@ -25,9 +25,6 @@ from .task_result import TaskResult, TaskStatus
# API DataSource
from .api import ApiClient, ApiDataSource, BaseApiHandler, register as register_api_datasource
# DB base helpers
from .db import DbDataSource, BaseDbHandler
# JSONL DataSource
from .jsonl import JsonlDataSource, BaseJsonlHandler, register as register_jsonl_datasource
@@ -65,10 +62,6 @@ __all__ = [
"ApiDataSource",
"BaseApiHandler",
# DB base helpers
"DbDataSource",
"BaseDbHandler",
# JSONL DataSource
"JsonlDataSource",
"BaseJsonlHandler",
sync_state_machine/datasource/api/datasource.py
+5 -6
View File
@@ -12,7 +12,7 @@ import json
from typing import Optional
from pathlib import Path
from ..datasource import BaseDataSource
from ..handler import NodeHandler, ensure_handler_compat
from ..handler import NodeHandler
from ..task_result import TaskResult
from .client import ApiClient
@@ -72,16 +72,15 @@ class ApiDataSource(BaseDataSource):
Args:
handler: API Handler 实例
"""
compatible_handler = ensure_handler_compat(handler)
compatible_handler.set_api_client(self.client)
compatible_handler.set_poll_mode(self.poll_mode)
handler.set_api_client(self.client)
handler.set_poll_mode(self.poll_mode)
# 调用基类注册方法
super().register_handler(compatible_handler)
super().register_handler(handler)
# 如果 collection 已设置，确保新 handler 拿到引用
if self._collection is not None:
compatible_handler.set_collection(self._collection) # type: ignore[arg-type]
handler.set_collection(self._collection) # type: ignore[arg-type]
@staticmethod
def _fmt_trace_value(value, max_len: int = 600) -> str:
sync_state_machine/datasource/datasource.py
+2 -4
View File
@@ -31,7 +31,6 @@ if TYPE_CHECKING:
from ..common.types import SyncStatus
from ..common.type_safety import get_pydantic_model_fields
from .handler import ensure_handler_compat
from .task_result import TaskResult
from ..engine import (
StateMachineConfig,
@@ -379,9 +378,8 @@ class BaseDataSource(ABC):
示例：
datasource.register_handler(ContractNodeHandler(api_client))
"""
compatible_handler = ensure_handler_compat(handler)
self._handlers[compatible_handler.node_type] = compatible_handler
self._apply_target_project_ids_to_handler(compatible_handler)
self._handlers[handler.node_type] = handler
self._apply_target_project_ids_to_handler(handler)
def set_target_project_ids(self, target_project_ids: Optional[List[str]]) -> None:
"""设置项目过滤列表，并在注册阶段下发给所有 handler。"""
sync_state_machine/datasource/db/__init__.py
-8
View File
@@ -1,8 +0,0 @@
from __future__ import annotations
"""DB DataSource 基类模块。"""
from .datasource import DbDataSource
from .handler import BaseDbHandler
__all__ = ["DbDataSource", "BaseDbHandler"]
sync_state_machine/datasource/db/datasource.py
-22
View File
@@ -1,22 +0,0 @@
"""DbDataSource - 数据库数据源协调层。"""
from __future__ import annotations
from ..datasource import BaseDataSource
class DbDataSource(BaseDataSource):
"""基于数据库 Handler 的数据源实现。"""
async def close(self) -> None:
return None
def _on_node_loaded(self, handler, node) -> None:
node.append_log(
f"DB链路[load] node_type={node.node_type} data_id={node.data_id or '<empty>'}"
)
def _on_in_progress_result(self, handler, node, result) -> None:
node.append_log(
f"DB链路[submit] action={node.action.value} data_id={node.data_id or '<empty>'}"
)
sync_state_machine/datasource/db/handler.py
-68
View File
@@ -1,68 +0,0 @@
"""BaseDbHandler - DB Handler 基类。"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Set, TypeVar, TYPE_CHECKING
from pydantic import BaseModel
from ...common.type_safety import get_pydantic_model_fields
from ..handler import BaseNodeHandler
if TYPE_CHECKING:
from ...common.sync_node import SyncNode
from .datasource import DbDataSource
T = TypeVar("T", bound=BaseModel)
class BaseDbHandler(BaseNodeHandler[T]):
"""数据库 Handler 基类。"""
def __init__(
self,
datasource: "DbDataSource",
node_type: str,
node_class: type["SyncNode[T]"],
schema: type[T],
):
super().__init__(node_type, node_class, schema)
self.datasource = datasource
self.update_schemas: List[type] = []
self._target_project_ids: Optional[Set[str]] = None
def set_target_project_ids(self, target_project_ids: List[str]) -> None:
if target_project_ids:
self._target_project_ids = {str(item) for item in target_project_ids if str(item)}
else:
self._target_project_ids = None
def get_update_fields(self, *, exclude: frozenset = frozenset({"id"})) -> List[str]:
fields: set[str] = set()
for schema in self.update_schemas:
for field_name in get_pydantic_model_fields(schema):
if field_name not in exclude:
fields.add(field_name)
return sorted(fields)
def _create_basic_node(self, data: Dict[str, Any], depend_ids: Optional[List[str]] = None) -> "SyncNode[T]":
data_id = self.extract_id(data)
if self._collection is None:
raise RuntimeError("Collection not set. Call set_collection() before creating nodes.")
if data_id:
existing_node = self._collection.get_by_data_id(self.node_type, data_id)
if existing_node is not None:
existing_node.depend_ids = list(depend_ids or [])
existing_node.set_data(data.copy())
existing_node.set_origin_data(data.copy())
return existing_node
node_id = self._collection.generate_node_id()
return self.node_class(
node_id=node_id,
data_id=data_id,
data=data.copy(),
depend_ids=depend_ids or [],
origin_data=data.copy(),
)
sync_state_machine/datasource/handler.py
-81
View File
@@ -479,87 +479,6 @@ class BaseNodeHandler(ABC, Generic[T]):
return self.create_node(data)
class CompatibleNodeHandler(Generic[T]):
"""兼容旧式 Handler，实现扩展后的可选接口并委托给原对象。"""
def __init__(self, handler: NodeHandler[T]):
self._handler = handler
@property
def node_type(self) -> str:
return self._handler.node_type
@property
def node_class(self) -> Type["SyncNode[T]"]:
return self._handler.node_class
@property
def schema(self) -> Type[T]:
return self._handler.schema
def set_collection(self, collection: "DataCollection") -> None:
self._handler.set_collection(collection)
def set_handler_config(self, config: Dict[str, Any]) -> None:
method = getattr(self._handler, "set_handler_config", None)
if callable(method):
method(config)
def set_target_project_ids(self, target_project_ids: List[str]) -> None:
method = getattr(self._handler, "set_target_project_ids", None)
if callable(method):
method(target_project_ids)
def set_api_client(self, client: "ApiClient") -> None:
method = getattr(self._handler, "set_api_client", None)
if callable(method):
method(client)
def set_poll_mode(self, mode: str) -> None:
method = getattr(self._handler, "set_poll_mode", None)
if callable(method):
method(mode)
def get_update_fields(self, *, exclude: frozenset[str] = frozenset({"id"})) -> List[str]:
method = getattr(self._handler, "get_update_fields", None)
if callable(method):
return method(exclude=exclude)
return []
async def load(self) -> List["SyncNode"]:
return await self._handler.load()
async def create_all(self, nodes: List["SyncNode"]) -> List[TaskResult]:
return await self._handler.create_all(nodes)
async def update_all(self, nodes: List["SyncNode"]) -> List[TaskResult]:
return await self._handler.update_all(nodes)
async def delete_all(self, nodes: List["SyncNode"]) -> List[TaskResult]:
return await self._handler.delete_all(nodes)
async def poll_tasks(self, task_ids: List[str]) -> Dict[str, TaskResult]:
return await self._handler.poll_tasks(task_ids)
async def validate(self, data: Dict[str, Any]) -> ValidationResult:
return await self._handler.validate(data)
def extract_id(self, data: Dict[str, Any]) -> str:
return self._handler.extract_id(data)
def _create_node(self, data: Dict[str, Any]) -> "SyncNode[T]":
return self._handler._create_node(data)
def __getattr__(self, item: str) -> Any:
return getattr(self._handler, item)
def ensure_handler_compat(handler: NodeHandler[T]) -> NodeHandler[T]:
if isinstance(handler, BaseNodeHandler) or isinstance(handler, CompatibleNodeHandler):
return handler
return CompatibleNodeHandler(handler)
class NodeHandlerRegistry:
"""节点处理器注册表"""
sync_state_machine/pipeline/__init__.py
-2
View File
@@ -14,7 +14,6 @@ from ..config import (
build_default_config,
build_config,
apply_config_overrides,
apply_env_overrides,
config_snapshot,
)
from .factory import (
@@ -46,6 +45,5 @@ __all__ = [
"build_default_config",
"build_config",
"apply_config_overrides",
"apply_env_overrides",
"config_snapshot",
]
tests/unit/test_datasource_type_registry.py
+10 -36
View File
@@ -18,7 +18,7 @@ from sync_state_machine.config import (
)
from sync_state_machine.datasource import BaseDataSource
from sync_state_machine.datasource import ensure_builtin_datasource_types_registered
from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
from sync_state_machine.datasource.handler import BaseNodeHandler
from sync_state_machine.pipeline.factory import create_pipeline_from_config
from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
@@ -56,11 +56,12 @@ class MemoryDataSource(BaseDataSource):
self.namespace = namespace
class MemoryHandler(BaseDbHandler[MemorySchema]):
class MemoryHandler(BaseNodeHandler[MemorySchema]):
__test__ = False
def __init__(self, datasource: MemoryDataSource):
super().__init__(datasource, "memory_demo", MemoryNode, MemorySchema)
super().__init__("memory_demo", MemoryNode, MemorySchema)
self.datasource = datasource
def extract_id(self, data: dict[str, str]) -> str:
return str(data.get("id", ""))
@@ -75,11 +76,12 @@ class MemoryHandler(BaseDbHandler[MemorySchema]):
return []
class MemoryRemoteDbHandler(BaseDbHandler[MemorySchema]):
class MemoryRemoteDbHandler(BaseNodeHandler[MemorySchema]):
__test__ = False
def __init__(self, datasource: DbDataSource):
super().__init__(datasource, "memory_demo", MemoryNode, MemorySchema)
def __init__(self, datasource: BaseDataSource):
super().__init__("memory_demo", MemoryNode, MemorySchema)
self.datasource = datasource
def extract_id(self, data: dict[str, str]) -> str:
return str(data.get("id", ""))
@@ -152,36 +154,8 @@ async def test_create_pipeline_from_config_supports_registered_custom_datasource
await pipeline.close()
def test_build_config_supports_same_file_datasource_bootstrap(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
bootstrap_module = tmp_path / "memory_bootstrap_demo.py"
bootstrap_module.write_text(
"from pydantic import ConfigDict\n"
"from sync_state_machine.config import BaseDataSourceConfig, register_datasource_type\n"
"from sync_state_machine.datasource import BaseDataSource\n"
"\n"
"class BootstrapMemoryConfig(BaseDataSourceConfig):\n"
" model_config = ConfigDict(extra=\"forbid\", validate_assignment=True)\n"
" type: str = \"memory_bootstrap\"\n"
" namespace: str = \"demo\"\n"
"\n"
"class BootstrapMemoryDataSource(BaseDataSource):\n"
" def __init__(self, namespace: str):\n"
" super().__init__()\n"
" self.namespace = namespace\n"
"\n"
"def build_memory_datasource(config, endpoint_name):\n"
" return BootstrapMemoryDataSource(namespace=f\"{endpoint_name}:{config.namespace}\")\n"
"\n"
"def register():\n"
" register_datasource_type(\n"
" \"memory_bootstrap\",\n"
" config_model=BootstrapMemoryConfig,\n"
" factory=build_memory_datasource,\n"
" replace=True,\n"
" )\n",
encoding="utf-8",
)
monkeypatch.syspath_prepend(str(tmp_path))
def test_build_config_supports_same_file_datasource_bootstrap(tmp_path: Path) -> None:
register_datasource_type("memory_bootstrap", config_model=MemoryConfig, factory=_memory_factory, replace=True)
profile_path = tmp_path / "custom-memory.yaml"
profile_path.write_text(
tests/unit/test_factory_db_handler_registration.py
+8 -6
View File
@@ -14,7 +14,8 @@ from sync_state_machine.config import (
PipelineRunConfig,
register_datasource_type,
)
from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
from sync_state_machine.datasource import BaseDataSource
from sync_state_machine.datasource.handler import BaseNodeHandler
from sync_state_machine.datasource.jsonl import BaseJsonlHandler, JsonlDataSource
from sync_state_machine.pipeline.factory import _register_handlers
from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
@@ -42,11 +43,12 @@ class TestDbStrategy(DefaultSyncStrategy[TestDbSchema]):
schema = TestDbSchema
class TestDbHandler(BaseDbHandler[TestDbSchema]):
class TestDbHandler(BaseNodeHandler[TestDbSchema]):
__test__ = False
def __init__(self, datasource: DbDataSource):
super().__init__(datasource, "test_db_handler", TestDbNode, TestDbSchema)
def __init__(self, datasource: BaseDataSource):
super().__init__("test_db_handler", TestDbNode, TestDbSchema)
self.datasource = datasource
def extract_id(self, data: dict[str, str]) -> str:
return str(data.get("id", ""))
@@ -72,7 +74,7 @@ def _register_test_db_datasource_type() -> None:
register_datasource_type(
"test_db",
config_model=TestRegisteredDbConfig,
factory=lambda config, endpoint_name: DbDataSource(),
factory=lambda config, endpoint_name: BaseDataSource(),
replace=True,
)
@@ -103,7 +105,7 @@ def test_register_handlers_uses_db_handler_for_db_datasource(tmp_path: Path) ->
logging=LoggingConfig(initialize=False),
)
local_ds = DbDataSource()
local_ds = BaseDataSource()
remote_ds = JsonlDataSource(tmp_path)
_register_handlers(config, local_ds, remote_ds, ["test_db_handler"])
tests/unit/test_factory_endpoint_initialization.py
+7 -5
View File
@@ -14,7 +14,8 @@ from sync_state_machine.config import (
PipelineRunConfig,
register_datasource_type,
)
from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
from sync_state_machine.datasource import BaseDataSource
from sync_state_machine.datasource.handler import BaseNodeHandler
from sync_state_machine.pipeline.factory import create_pipeline_from_config
from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
@@ -41,11 +42,12 @@ class EndpointInitStrategy(DefaultSyncStrategy[EndpointInitSchema]):
schema = EndpointInitSchema
class EndpointInitDbHandler(BaseDbHandler[EndpointInitSchema]):
class EndpointInitDbHandler(BaseNodeHandler[EndpointInitSchema]):
__test__ = False
def __init__(self, datasource: DbDataSource):
super().__init__(datasource, "factory_endpoint_demo", EndpointInitNode, EndpointInitSchema)
def __init__(self, datasource: BaseDataSource):
super().__init__("factory_endpoint_demo", EndpointInitNode, EndpointInitSchema)
self.datasource = datasource
def extract_id(self, data: dict[str, str]) -> str:
return str(data.get("id", ""))
@@ -85,7 +87,7 @@ def _register_endpoint_init_datasource_type() -> None:
register_datasource_type(
"endpoint_init_db",
config_model=EndpointInitDbConfig,
factory=lambda config, endpoint_name: DbDataSource(),
factory=lambda config, endpoint_name: BaseDataSource(),
replace=True,
)
tests/unit/test_run_presets_yaml_loader.py
+29 -30
View File
@@ -4,8 +4,11 @@ from pathlib import Path
import pytest
from sync_state_machine.config import build_config_from_file
from pydantic import ConfigDict
from sync_state_machine.config import BaseDataSourceConfig, build_config_from_file, register_datasource_type
from sync_state_machine.config.run_presets import load_overrides_from_file
from sync_state_machine.datasource import BaseDataSource
def test_load_overrides_from_yaml_file(tmp_path: Path) -> None:
@@ -32,7 +35,7 @@ local_datasource:
assert str(tmp_path) in overrides["local_datasource"]["jsonl_dir"]
def test_load_overrides_executes_generic_extensions_bootstraps(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
def test_load_overrides_ignores_extensions_bootstraps(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
bootstrap_module = tmp_path / "profile_bootstrap_demo.py"
bootstrap_module.write_text(
"BOOTSTRAP_CALLED = False\n"
@@ -68,37 +71,33 @@ logging:
import profile_bootstrap_demo
assert profile_bootstrap_demo.BOOTSTRAP_CALLED is True
assert profile_bootstrap_demo.BOOTSTRAP_CALLED is False
def test_build_config_from_file_uses_registered_bootstraps(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
bootstrap_module = tmp_path / "config_from_file_bootstrap.py"
bootstrap_module.write_text(
"from pydantic import ConfigDict\n"
"from sync_state_machine.config import BaseDataSourceConfig, register_datasource_type\n"
"from sync_state_machine.datasource import BaseDataSource\n"
"\n"
"class FileBackedConfig(BaseDataSourceConfig):\n"
" model_config = ConfigDict(extra=\"forbid\", validate_assignment=True)\n"
" type: str = \"file_backed\"\n"
" namespace: str = \"demo\"\n"
"\n"
"class FileBackedDatasource(BaseDataSource):\n"
" pass\n"
"\n"
"def build_datasource(config, endpoint_name):\n"
" return FileBackedDatasource()\n"
"\n"
"def register():\n"
" register_datasource_type(\n"
" \"file_backed\",\n"
" config_model=FileBackedConfig,\n"
" factory=build_datasource,\n"
" replace=True,\n"
" )\n",
encoding="utf-8",
class FileBackedConfig(BaseDataSourceConfig):
model_config = ConfigDict(extra="forbid", validate_assignment=True)
type: str = "file_backed"
namespace: str = "demo"
class FileBackedDatasource(BaseDataSource):
def __init__(self, namespace: str):
super().__init__()
self.namespace = namespace
def _build_file_backed_datasource(config, endpoint_name):
return FileBackedDatasource(namespace=f"{endpoint_name}:{config.namespace}")
def test_build_config_from_file_uses_registered_datasource_type(tmp_path: Path) -> None:
register_datasource_type(
"file_backed",
config_model=FileBackedConfig,
factory=_build_file_backed_datasource,
replace=True,
)
monkeypatch.syspath_prepend(str(tmp_path))
profile = tmp_path / "profile.yaml"
profile.write_text(