From ba73b30a04a2fb91790bfbdd4d957dcb622c2853 Mon Sep 17 00:00:00 2001 From: strepsiades Date: Tue, 7 Apr 2026 20:22:25 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0sqlalchemy=E5=90=8E=E7=AB=AF?= =?UTF-8?q?=EF=BC=8C=E4=BC=98=E5=8C=96=E6=97=A5=E5=BF=97=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 3 +- docs/getting_started.md | 2 +- docs/runtime_config_guide.md | 621 ++++++++++++------ pyproject.toml | 2 + scripts/auto_test_lib.py | 67 +- sync_state_machine/common/__init__.py | 3 +- sync_state_machine/common/binding.py | 4 +- sync_state_machine/common/persistence.py | 168 +++-- .../common/persistence_backends/__init__.py | 3 +- .../sqlalchemy_backend.py | 428 ++++++++++++ .../persistence_backends/sqlite_backend.py | 5 +- .../common/sqlalchemy_backend.py | 3 + sync_state_machine/config/persist_config.py | 63 +- sync_state_machine/datasource/api/client.py | 82 ++- .../datasource/api/datasource.py | 25 + sync_state_machine/datasource/datasource.py | 4 +- sync_state_machine/logging.py | 25 +- sync_state_machine/pipeline/factory.py | 10 +- .../pipeline/multi_project_pipeline.py | 4 +- sync_state_machine/pipeline/summary_report.py | 55 +- tests/unit/test_api_client_rate_limit.py | 93 +++ ...t_persist_config_and_sqlalchemy_backend.py | 147 +++++ tests/unit/test_summary_report.py | 77 ++- 23 files changed, 1586 insertions(+), 308 deletions(-) create mode 100644 sync_state_machine/common/persistence_backends/sqlalchemy_backend.py create mode 100644 sync_state_machine/common/sqlalchemy_backend.py create mode 100644 tests/unit/test_persist_config_and_sqlalchemy_backend.py diff --git a/README.md b/README.md index 5738331..f97ec1b 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,7 @@ ecm-sync-run --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml - 仓库内直接运行：`python run.py --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml` - 已安装命令行入口：`ecm-sync-run --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml` - 多项目 JSONL 示例：`python run.py --config_path=run_profiles/preset/jsonl_to_jsonl.multi_project.default.yaml` +- 当前已经支持多 project 运行；标准写法是直接使用多项目编排配置文件，或在支持 implicit multi-project 的 profile 中由配置本身决定运行模式。 - 自定义配置：从 `run_profiles/preset/` 复制一份到 `run_profiles/*.local.yaml`，只改有差异的字段即可。 ### API 同步 @@ -144,7 +145,7 @@ ecm-sync-run --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml - 第一次接手仓库：`docs/getting_started.md` - 想理解整体分层：`docs/architecture.md` - 想理解状态与迁移：`docs/node_state_definition.md` + `docs/state_machine.md` -- 想调运行配置：`docs/runtime_config_guide.md` + `run_profiles/README.md` +- 想调运行配置：`docs/runtime_config_guide.md` + `run_profiles/README.md`（包含多项目配置写法） - 想接新数据源或改 handler：`sync_state_machine/datasource/README.md` + `docs/library_integration_guide.md` - 想排障：`docs/operations.md` - 想补测试或跑联调：`docs/testing_guide.md` + `tests/README.md` diff --git a/docs/getting_started.md b/docs/getting_started.md index 70b50c5..3dc1698 100644 --- a/docs/getting_started.md +++ b/docs/getting_started.md @@ -46,7 +46,7 @@ pytest tests/integration/test_toolchain_config_and_graph.py -q - 想知道系统整体怎么分层：`docs/architecture.md` - 想知道状态机怎么工作：`docs/node_state_definition.md` 和 `docs/state_machine.md` - 想知道一次同步是怎么跑的：`docs/sync_flow.md` -- 想调运行配置：`docs/runtime_config_guide.md` +- 想调运行配置：`docs/runtime_config_guide.md`（包含单项目、多项目、`persist.url`、`sqlalchemy`、业务侧注册 datasource 的当前写法） - 想接新 datasource 或 handler：`sync_state_machine/datasource/README.md` 和 `docs/library_integration_guide.md` - 想看测试体系：`docs/testing_guide.md` 和 `tests/README.md` - 想做排障：`docs/operations.md` diff --git a/docs/runtime_config_guide.md b/docs/runtime_config_guide.md index c7177d1..ae2edfb 100644 --- a/docs/runtime_config_guide.md +++ b/docs/runtime_config_guide.md @@ -1,299 +1,508 @@ -# 运行配置业务说明 +# 运行配置写法说明 -本文档从业务视角说明配置字段含义、对阶段行为的影响，以及推荐的覆盖方式。 +这份文档只讲一件事：现在 `ecm_sync_system` 的配置文件应该怎么写。 -## 0. 配置约定 +重点不是把所有默认值抄一遍，而是说明当前真实生效的写法、哪些字段是内置能力、哪些字段来自业务侧扩展，以及单项目和多项目文件各自长什么样。 -当前运行配置分成两层： +## 1. 先分清两类配置文件 -1. 输入配置 - - 你写在 `run_profiles/*.yaml` 或 `run_profiles/preset/*.default.yaml` 里的内容。 - - 只建议写必要覆盖项，不要把默认值全部展开。 -2. Resolved Full Config - - pipeline 启动时打印并写入日志的全量配置。 - - 其中已经补齐 datasource 默认值、策略默认值和日志/持久化默认值。 +当前有两类 YAML： -也就是说： -- 模板应当尽量短。 -- 排障时看启动日志里的 **Resolved Full Config**。 +1. 单 pipeline 配置 + - 对应 `PipelineRunConfig` + - 用来描述一条完整同步链路 + - 顶层通常包含：`node_types`、`local_datasource`、`remote_datasource`、`persist`、`logging`、`strategies` +2. 多项目编排配置 + - 对应 `MultiProjectRunConfig` + - 只负责描述“先跑全局，再按项目并发跑哪些项目” + - 顶层包含：`global_config`、`default_project_config`、`max_concurrency`、`project_bootstrap_node_types`、`projects` -### 0.1 一个最小输入示例 +判断规则很简单： + +- 如果文件里有 `global_config`、`default_project_config`、`projects`，它就是多项目编排配置。 +- 否则按单 pipeline 配置解析。 + +## 2. 配置加载约定 + +当前约定如下： + +- 跟踪在仓库里的模板和示例，只写必要覆盖项，不重复展开默认值。 +- pipeline 启动后会打印一份 `Resolved Full Config`，它才是最终生效的完整配置。 +- 排障优先看 `Resolved Full Config`，不要只盯着输入 YAML。 + +## 3. 占位符规则 + +`ecm_sync_system` 核心库内置只认识一个占位符： + +- `${PROJECT_ROOT}`：运行时替换为 `ecm_sync_system` 仓库根目录绝对路径。 + +例如： ```yaml +persist: + backend: sqlite + db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/demo.db +``` + +需要特别区分的是： + +- `${BACKEND_ROOT}`、`${ECM_SYNC_LOG_DIR}` 这类占位符不是核心库通用语义。 +- 它们通常是 backend 集成脚本在读取 profile 时额外替换的调用方语义。 +- 如果你不是通过 backend 的联调脚本运行，而是直接在 `ecm_sync_system` 内跑 profile，就不要假设这些占位符一定可用。 + +## 4. 单 pipeline 配置的当前写法 + +### 4.1 最小骨架 + +```yaml +node_types: + - company + - supplier + - project + local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered + remote_datasource: - type: jsonl - jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource + type: api + api_base_url: http://127.0.0.1:8000/api/third/v2 + api_uid: your_uid + api_secret: your_secret + persist: - db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/demo.db backend: sqlite - backend_options: {} + db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/demo.db + +logging: + file_dir: ${PROJECT_ROOT}/test_results/sync_state_machine + file_name_pattern: demo_{timestamp}.log + strategies: project: config: auto_bind: true auto_bind_fields: - name + local_orphan_action: none + remote_orphan_action: none + update_direction: push ``` -### 0.2 对应的全量 resolved 结果骨架 +### 4.2 顶层字段含义 + +- `node_types` + - 本轮参与同步的节点类型列表。 + - 顺序会影响执行顺序，依赖链通常要把上游类型放前面。 +- `local_datasource` / `remote_datasource` + - 定义两侧数据源类型和对应参数。 +- `persist` + - 定义持久化后端及目标位置。 +- `logging` + - 定义日志初始化、落盘位置和级别。 +- `strategies` + - 定义每个 `node_type` 的同步策略。 + +## 5. datasource 字段怎么写 + +### 5.1 内置 datasource type + +核心库内置注册了两种 datasource type： + +1. `jsonl` +2. `api` + +#### `jsonl` 写法 ```yaml -node_types: - - company - - supplier - - user - - project - ... local_datasource: type: jsonl - jsonl_dir: /abs/path/filtered_datasource/datasource/filtered + jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered read_only: false - handler_configs: {} + handler_configs: + project: + data_id_filter: + - project-id-1 +``` + +字段说明： + +- `jsonl_dir`：JSONL 数据目录。 +- `read_only`：是否只读。 +- `handler_configs`：按 `node_type` 下发给 handler 的扩展参数。 + +#### `api` 写法 + +```yaml remote_datasource: - type: jsonl - jsonl_dir: /abs/path/remote_datasource/datasource - read_only: false - handler_configs: {} + type: api + api_base_url: http://127.0.0.1:8000/api/third/v2 + api_uid: your_uid + api_secret: your_secret + api_debug: true + poll_max_retries: 10 + poll_interval: 0.5 + api_rate_limit_max_requests: 30 + api_rate_limit_window_seconds: 10.0 + handler_configs: + project: + data_id_filter: + - remote-project-id-1 +``` + +字段说明： + +- `api_debug`：开启后输出更详细的请求日志。 +- `poll_max_retries` / `poll_interval`：异步任务轮询控制。 +- `api_rate_limit_max_requests` / `api_rate_limit_window_seconds`：全局限流窗口。 +- `handler_configs`：同样按 `node_type` 下发业务参数。 + +### 5.2 业务侧注册的 datasource type + +核心库允许业务系统在启动阶段注册自己的 datasource type。 + +因此，配置里不只可以写 `jsonl` 和 `api`，还可以写业务侧注册出来的类型，例如 backend 里的 `depm_db`。 + +当前 backend 的 `depm_db` 真实写法是： + +```yaml +local_datasource: + type: depm_db + tenant_id: "019976df-0628-7c46-a896-81d3f290a7df" + app: + config_file: ${BACKEND_ROOT}/config.yaml + overrides: + database: + url: mysql+aiomysql://root:123456@127.0.0.1/depm_prod_local_auto_test?charset=utf8mb4 + handler_configs: + project: + data_id_filter: + - 019a62a5-2323-7be2-99c8-82534f8befa4 +``` + +这个例子说明两点： + +1. datasource 的可写字段由对应注册模型决定，不是所有 type 都共用同一套字段。 +2. 如果 `type` 没有先被注册，配置校验会直接报错，不会做降级处理。 + +`depm_db` 这类业务扩展字段需要以业务侧实现为准。对 backend 当前实现来说： + +- `tenant_id`：DEPM 本地 datasource 的租户过滤 ID。 +- `app.config_file`：backend 配置文件路径，默认可落回 `backend/config.yaml`。 +- `app.overrides`：直接传给 backend `Settings()` 的覆盖项。 + +## 6. strategy 写法 + +### 6.1 当前推荐写法 + +`strategies` 现在推荐写成“按 `node_type` 为键”的映射，而不是手工写一长串列表。 + +```yaml strategies: - - node_type: project - skip_sync: false + supplier: + skip_post_check: true config: auto_bind: true auto_bind_fields: - - name - pre_bind_data_id: [] + - credit_code + depend_fields: {} + local_orphan_action: none + remote_orphan_action: create_local + update_direction: none + project: + domain_option: + pull_group_plan_production_date: true + config: + auto_bind: false + auto_bind_fields: [] + pre_bind_data_id: + - local_data_id: local-project-id + remote_data_id: remote-project-id depend_fields: company_id: company - local_orphan_action: create_remote + local_orphan_action: none remote_orphan_action: none update_direction: push - post_check_ignore_fields: [] - compare_ignore_fields: [] - domain_option: {} +``` + +### 6.2 常用字段 + +- `auto_bind` + - 是否开启自动绑定。 +- `auto_bind_fields` + - 自动绑定使用的业务键字段。 +- `pre_bind_data_id` + - 显式 local/remote data_id 绑定对。 +- `depend_fields` + - 依赖字段到依赖 `node_type` 的映射。 +- `local_orphan_action` / `remote_orphan_action` + - 孤儿节点创建策略。 +- `update_direction` + - `push` / `pull` / `none`。 +- `post_check_ignore_fields` + - 只影响最终一致性检查，不影响 create/update 差异检测。 +- `compare_ignore_fields` + - schema diff / validate 时忽略的顶层字段。 +- `skip_sync` + - 只跳过 create/update，不跳过 load/bind。 +- `skip_post_check` + - 跳过该类型的 post-check reload 与一致性校验。 +- `domain_option` + - 领域策略自己的扩展参数，由对应 domain 自己解释。 + +### 6.3 `preset` 仍然可用 + +如果只是想套一组预设策略，可以这样写： + +```yaml +strategies: + project: + preset: pull_only +``` + +当前内置预设包括： + +- `first_sync` +- `daily_sync` +- `repair_sync` +- `pull_only` +- `push_only` + +### 6.4 已废弃的旧字段不要再写 + +以下 legacy key 已明确拒绝： + +- `force_sync` +- `load_mode` + +如果你在 strategy 里写这两个字段，配置会直接报错。 + +当前替代方式是： + +- 是否跳过同步，用 `skip_sync` +- handler 特有加载参数，写到 `datasource.handler_configs.` + +## 7. persist 的当前写法 + +### 7.1 校验规则 + +当前 `persist` 配置遵循下面的规则： + +1. `db_path` 和 `url` 至少要有一个。 +2. 当 `backend != sqlite` 时，必须显式提供 `url`。 +3. `wipe_on_start=true` 只会尝试删除可映射到文件路径的持久化文件；对 MySQL 这类远程数据库不会帮你删库。 + +### 7.2 SQLite 文件持久化 + +```yaml persist: backend: sqlite - db_path: /abs/path/test_results/sync_state_machine/demo.db - backend_options: {} + db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/simple_pipeline_sm.db enable: true wipe_on_start: false +``` + +### 7.3 SQLAlchemy 持久化到 SQLite URL + +```yaml +persist: + backend: sqlalchemy + url: sqlite+aiosqlite:////abs/path/to/pipeline.db + enable: true + wipe_on_start: false +``` + +### 7.4 SQLAlchemy 持久化到 MySQL + +```yaml +persist: + backend: sqlalchemy + url: mysql+aiomysql://root:123456@127.0.0.1/ecm_sync_system?charset=utf8mb4 + enable: true + wipe_on_start: false +``` + +当前语义： + +- `sqlalchemy` backend 会自动创建它自己的 `nodes` 和 `bindings` 表及索引。 +- 它不会替你创建 MySQL 数据库本身，所以 `ecm_sync_system` 这个库需要提前存在。 +- 现有表结构已经兼容 MySQL，不需要再额外给主键字段手工指定前缀长度。 + +### 7.5 `backend_options` 什么时候用 + +`backend_options` 仍然保留，但现在优先使用显式字段： + +- `db_path` +- `url` +- `enable` +- `wipe_on_start` + +除非某个自定义 backend 明确要求，否则不要把主连接信息再塞进 `backend_options`。 + +## 8. logging 的当前写法 + +```yaml logging: initialize: true - file_path: null - file_dir: null - file_name_pattern: null + file_dir: ${PROJECT_ROOT}/test_results/sync_state_machine + file_name_pattern: simple_pipeline_sm_{timestamp}.log level: 20 suppress_node_logs: true ``` -## 1. 推荐使用方式（先 default，再覆盖） +字段说明： -1. 先使用基线配置运行： - - `run_profiles/preset/jsonl_to_jsonl.default.yaml` -2. 若需项目定制，再复制为： - - `run_profiles/jsonl_to_jsonl.local.yaml` -3. 仅改有业务差异的字段，避免一次改太多导致排障困难。 +- `initialize` + - 是否初始化应用日志。 +- `file_path` + - 直接指定日志文件绝对路径。 +- `file_dir` + - 只指定目录时，实际文件名由 `file_name_pattern` 生成。 +- `file_name_pattern` + - 支持 `{timestamp}` 占位。 +- `level` + - 默认 `20`，即 `INFO`。 +- `suppress_node_logs` + - 是否压缩逐节点日志噪音。 -占位符： -- `${PROJECT_ROOT}` 会在运行时替换为项目根目录绝对路径。 +## 9. 多项目编排文件怎么写 -## 2. 顶层字段（你在改什么） +### 9.1 结构 -- `node_types` - - 本轮参与同步的业务类型列表。 - - 顺序会影响依赖链执行时机（当前按 node_type 串行执行 bind/create/update）。 -- `local_datasource` / `remote_datasource` - - 分别定义两侧数据来源与执行端。 -- `strategies` - - 每个业务类型的行为策略（依赖、自动绑定、创建方向、更新方向、跳过等）。 -- `persist` - - 持久化开关与数据库位置。 -- `persist.backend` - - 持久化后端注册表里的名称，例如 `sqlite`。 - - backend 实现会自行解释 `persist` 的其余字段。 -- `persist.backend_options` - - 后端私有配置，后端自己决定如何解释。 - -### 6. 持久化后端注册 - -如果你希望在 backend 初始化阶段替换持久化后端，可以先在启动代码里按名称注册，再通过配置选择，不需要改调用方。 - -#### 6.1 内置 SQLite +多项目文件不是把所有字段都堆在一起，而是引用两份单 pipeline 配置： ```yaml -persist: - backend: sqlite - db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/demo.db - backend_options: {} +global_config: run_profiles/jsonl_to_jsonl.multi_project.global.yaml +default_project_config: run_profiles/jsonl_to_jsonl.multi_project.project.yaml + +project_bootstrap_node_types: + - company + - supplier + +max_concurrency: 8 + +projects: + - local_project_id: local-project-1 + remote_project_id: remote-project-1 + - local_project_id: local-project-2 + remote_project_id: remote-project-2 ``` -#### 6.2 注册自定义后端 +字段说明： -```python -from sync_state_machine.common.persistence import register_persistence_backend +- `global_config` + - 全局公共节点要跑的单 pipeline 配置。 +- `default_project_config` + - 每个项目默认使用的单 pipeline 配置。 +- `project_bootstrap_node_types` + - 需要从 global scope 继承到项目 scope 的公共节点类型，例如 `company`、`supplier`。 +- `max_concurrency` + - 项目并发数上限，默认 `20`。 +- `projects` + - 项目映射清单。 -register_persistence_backend("mysql", lambda config: MySQLPersistenceBackend(config)) -``` +### 9.2 项目项支持单独切配置 -然后在配置里选择这个名字： +如果某个项目需要特殊策略，可以在项目项里覆盖 `config_path`： ```yaml -persist: - backend: mysql - db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/demo.db - backend_options: - url: mysql+aiomysql://user:pass@127.0.0.1/depm_sync?charset=utf8mb4 - pool_size: 5 +projects: + - local_project_id: local-project-1 + remote_project_id: remote-project-1 + - local_project_id: local-project-2 + remote_project_id: remote-project-2 + config_path: run_profiles/special_project.pipeline.yaml ``` -后端实现会收到整份 `PersistConfig`，自行决定是读取 `db_path`、`backend_options`，还是额外扩展别的字段。 -- `logging` - - 运行日志输出位置与级别。 +此时该项目不会再用 `default_project_config`，而是改用自己的单 pipeline 配置文件。 -## 3. 数据源字段（DataSource） +### 9.3 当前多项目 profile 的实际风格 -### 3.1 JSONL 数据源 +以当前 backend 两项目推送链路为例，常见组合是： -- `type=jsonl`：使用本地 JSONL 文件作为数据源。 -- `jsonl_dir`：JSONL 目录路径。 -- `read_only=false`：允许该侧执行 create/update 回写；若为 true，通常用于只读加载。 -- `handler_configs.project.data_id_filter=[]`：项目白名单；为空时不进行项目筛选。 -- `handler_configs`：按节点类型下发给各业务 handler 的扩展参数。 +- `local_datasource.type: depm_db` +- `remote_datasource.type: api` +- `persist.backend: sqlalchemy` +- `persist.url: mysql+aiomysql://...` +- `strategies.project.config.pre_bind_data_id` 明确写每个项目的 local/remote 映射 -### 3.2 API 数据源 +这类 profile 本质上仍然是“多项目编排文件 + 两份单 pipeline 文件”的组合，不要把它理解成第三种配置格式。 -- `type=api`：对接 HTTP API。 -- `api_base_url` / `api_uid` / `api_secret`：接口访问参数。 -- `api_debug`：开启后输出更多调试信息。 -- `poll_max_retries` / `poll_interval`：异步任务轮询次数与间隔。 -- `handler_configs.project.data_id_filter`：建议显式配置的项目白名单；至少需要指定一个项目ID，以避免加载过多远程数据。 -- `handler_configs`：业务 handler 扩展参数。 +## 10. 当前推荐的写法习惯 -## 4. 策略字段（Strategy） +### 10.1 只写必要覆盖项 -每个 node_type 的 **Resolved Full Config** 都会包含以下字段，但输入配置不需要全部显式写出。 +推荐： -### 4.1 `update_direction` +- 只写与你当前环境、项目过滤、持久化目标、策略差异直接相关的字段。 -- `push`：本地变更推送到远端。 -- `pull`：远端变更拉取到本地。 -- `none`：关闭 update 阶段。 +不推荐： -### 4.2 `local_orphan_action` / `remote_orphan_action` +- 把默认值全部展开复制一遍。 -- `create_remote`：本地孤儿创建到远端。 -- `create_local`：远端孤儿创建到本地。 -- `none`：不自动创建。 -- `delete_local` / `delete_remote`：当前主流程未作为默认路径，谨慎使用。 +原因很直接： -### 4.3 `auto_bind` / `auto_bind_fields` / `pre_bind_data_id` +- 默认值一旦调整，你的大段本地 YAML 会悄悄过时。 +- 排障时也更难看出你到底改了什么。 -- `auto_bind=true`：在 `S02(MISSING)` 上尝试自动匹配。 -- `auto_bind=false`：不会做自动匹配，也不会走自动创建识别路径；通常会进入阻断态等待人工处理。 -- `auto_bind_fields`：自动匹配使用的业务键字段。 -- `pre_bind_data_id`：显式 data_id 预绑定清单（通用能力），格式为： - - `- local_data_id: "..."` - ` remote_data_id: "..."` - - 规则：若两侧数据源都存在这对 data_id 对应节点，且当前无绑定记录，则直接绑定。 - - 不依赖 `auto_bind/auto_bind_fields`，即使 `auto_bind=false` 也会执行。 - - 不做业务键一对一匹配检查，按你给定的映射对优先执行。 +### 10.2 handler 特有参数放 datasource 里 -### 4.4 `depend_fields` - -- 定义依赖字段到依赖 node_type 的映射，例如：`company_id -> company`。 -- 依赖节点状态非 `NORMAL` 时，会在 bind 的依赖检查阶段进入 `DEPENDENCY_ERROR`。 - -### 4.5 `skip_sync` - -- 运行配置中已不建议使用 `skip_sync`。 -- 推荐使用显式组合： - - `local_orphan_action: none` - - `remote_orphan_action: none` - - `update_direction: none` -- `skip_sync=true` 的当前语义是： - - 仍会执行该 node_type 的 load / bind - - 只跳过 create / update 写入阶段 -- 适合“保留绑定与依赖可见性，但临时禁止写入”的场景。 -- `none/none/none` 是更推荐的输入风格，因为它表达的是“该类型不做创建、不做补本地、不做更新”。 -- 这组配置不会绕过前置 load / bind / 依赖检查；它只会把该类型变成无写入策略。 - -### 4.6 Handler 级参数（`handler_configs`） - -- 建议将 handler 特有参数放在 datasource 的 `handler_configs.` 下。 -- 示例： - - `remote_datasource.handler_configs.supplier.load_mode: persisted_only` -- `persisted_only` 表示只使用已持久化数据，不主动从外部拉新数据（具体以对应 handler 实现为准）。 - -### 4.7 `domain_option` - -- `domain_option` 用于承载 domain 自己识别的运行时扩展参数。 -- 通用策略行为仍然放在 `config` 下；只有领域特例才放到 `domain_option`。 - -示例 1：`project` 先从远端拉回两个特殊投产日期字段，再执行默认 push +比如： ```yaml -project: - domain_option: - pull_group_plan_production_date: true - config: - update_direction: push +remote_datasource: + type: api + handler_configs: + contract_settlement_detail: + load_method: aggregate ``` -示例 2：`project_detail_data` 用单个开关启用固定业务规则，把集团保供/爬坡/挑战产能计划按 pull 处理 +这类参数应该写在 `datasource.handler_configs.`，不要错写到 strategy 里。 + +### 10.3 项目过滤显式下沉到 handler + +例如： ```yaml -project_detail_data: - domain_option: - pull_group_production_plans: true - config: - update_direction: push +handler_configs: + project: + data_id_filter: + - project-id-1 + - project-id-2 ``` -含义： -- `project.domain_option.pull_group_plan_production_date=true` 时，`ic_plan_first_production_date` 和 `ic_plan_full_production_date` 会先 remote -> local，再执行剩余字段的常规更新。 -- `project_detail_data.domain_option.pull_group_production_plans=true` 时，固定的 `ensure_production`、`climb_production`、`challenge_production` 会先 remote -> local；其余 key 继续沿用 `config.update_direction`。 +不要在 pipeline 外层再做一层临时过滤逻辑。项目范围应尽量体现在配置里，并由对应 handler 解释。 -## 5. 阶段行为与配置关系 +### 10.4 看到报错先修配置，不要做降级 -- bind 阶段 - - `depend_fields` 决定是否做依赖校验。 - - `pre_bind_data_id` 先按显式 local/remote data_id 映射做预绑定。 - - `auto_bind/auto_bind_fields` 决定是否继续进行业务键自动绑定。 -- create 阶段 - - `local_orphan_action` / `remote_orphan_action` 决定孤儿是否创建。 -- update 阶段 - - `update_direction` 决定是否和向哪侧更新。 -- 写入后 reload（Pipeline 内置） - - 对 API / 需要重新拉取权威状态的数据源，create 或 update 成功后仍会按 node_type reload。 - - 对纯 JSONL -> JSONL pipeline，reload 会跳过，直接复用当前内存状态。 - - 目的：避免对离线 JSONL 数据做无意义重复扫描，同时保留 API 数据源的权威状态刷新。 -- 同步后一致性检查（Pipeline 内置） - - 每个 node_type 在本轮同步后会基于绑定对执行数据一致性校验（比较时忽略 `id/_id` 字段）。 - - 若发现差异，会在日志中打印差异样本，并在汇总表新增 `Consistent` 列展示 `Y` / `N(差异数)`。 -- 全局跳过 - - `local_orphan_action=none` + `remote_orphan_action=none` + `update_direction=none` - 可将该类型收敛为“只做 load/bind，不做 create/update 写入”的只读同步策略。 +当前系统对配置保持 fail-fast： -## 6. 覆盖配置建议 +- datasource type 未注册会报错 +- legacy strategy key 会报错 +- `persist` 缺失目标会报错 +- profile 结构不符合 schema 会报错 -- 先在 default 跑通一轮，再做 local 覆盖。 -- 每次只改一类问题相关字段（例如只改 `project` 的依赖与自动绑定）。 -- 覆盖后优先看 `sync_log` 是否出现预期状态迁移。 +这属于预期行为。优先修正配置，而不是在文档或调用侧加 fallback。 -## 7. 持久化与日志字段 +## 11. 参考文件 -### `persist` +如果你想直接对照真实样例，优先看这些文件： -- `enable`：是否开启持久化。 -- `db_path`：SQLite 路径。 -- `wipe_on_start`：启动是否清理历史持久化。 +- `run_profiles/jsonl_to_api.yaml` +- `run_profiles/jsonl_to_jsonl.multi_project.yaml` +- backend 中的 `local_tests/sync_system_tests/depm_to_ecm_bootstrap_pull_then_project_push.two_projects.sqlalchemy.pipeline.yaml` -### `logging` +如果你想看配置模型定义，直接看： -- `file_dir`：日志目录。 -- `file_name_pattern`：文件命名模板。 -- `level`：日志级别。 +- `sync_state_machine/config/pipeline_config.py` +- `sync_state_machine/config/datasource_config.py` +- `sync_state_machine/config/persist_config.py` +- `sync_state_machine/config/multi_project_config.py` +- `sync_state_machine/config/strategy_config.py` -排障建议：运行失败时优先看日志中的 `状态机流转` 与 `reason=`，再对照 `docs/state_machine.md` 与 `docs/node_state_definition.md` 理解含义。 +排障时优先顺序： + +1. 看输入 YAML。 +2. 看启动日志里的 `Resolved Full Config`。 +3. 看对应 datasource type 是否真的已经注册。 +4. 再看运行日志和状态机结果。 diff --git a/pyproject.toml b/pyproject.toml index fb00a81..8e14a95 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,8 @@ dependencies = [ "PyYAML>=6.0,<7", "httpx>=0.27,<1", "aiosqlite>=0.20,<1", + "SQLAlchemy>=2.0,<3", + "greenlet>=3,<4", "pytz>=2024.1,<2027", "fastapi>=0.115,<1", "uvicorn>=0.30,<1", diff --git a/scripts/auto_test_lib.py b/scripts/auto_test_lib.py index 46d01da..0d051e6 100644 --- a/scripts/auto_test_lib.py +++ b/scripts/auto_test_lib.py @@ -15,6 +15,8 @@ from urllib import error, request import yaml +from sync_state_machine.common.persistence import PersistenceBackend + PROJECT_ROOT = Path(__file__).resolve().parents[1] @@ -312,26 +314,69 @@ def sqlite_rows(db_path: str | Path, sql: str, params: tuple[Any, ...] = ()) -> return [{key: row[key] for key in row.keys()} for row in rows] -def persistence_summary(db_path: str | Path, node_type: str | None = None) -> dict[str, Any]: +def persistence_rows( + *, + backend: str, + sql: str, + db_path: str | Path | None = None, + url: str | None = None, + limit: int = 200, +) -> list[dict[str, Any]]: + _, rows = PersistenceBackend.query_records( + backend=backend, + db_path=str(db_path) if db_path is not None else None, + url=url, + sql=sql, + limit=limit, + ) + return rows + + +def persistence_summary( + db_path: str | Path | None, + node_type: str | None = None, + *, + backend: str = "sqlite", + url: str | None = None, +) -> dict[str, Any]: + records = PersistenceBackend.records_summary( + backend=backend, + db_path=str(db_path) if db_path is not None else None, + url=url, + ) summary: dict[str, Any] = { + "backend": backend, "db_path": str(db_path), - "db_exists": Path(db_path).exists(), + "url": url, + "db_exists": records.get("db_exists", False), "bindings": [], "node_status": [], } - if not Path(db_path).exists(): + if not summary["db_exists"]: return summary + target_db_path = str(db_path) if db_path is not None else None if node_type: - summary["bindings"] = sqlite_rows( - db_path, - "SELECT node_type, local_id, remote_id, last_updated FROM bindings WHERE node_type = ? ORDER BY last_updated DESC LIMIT 20", - (node_type,), + escaped_node_type = node_type.replace("'", "''") + summary["bindings"] = persistence_rows( + backend=backend, + db_path=target_db_path, + url=url, + sql=( + "SELECT node_type, local_id, remote_id, last_updated " + f"FROM bindings WHERE node_type = '{escaped_node_type}' ORDER BY last_updated DESC LIMIT 20" + ), ) - summary["node_status"] = sqlite_rows( - db_path, - "SELECT collection_id, node_type, action, status, binding_status, COUNT(1) AS count FROM nodes WHERE node_type = ? GROUP BY collection_id, node_type, action, status, binding_status ORDER BY collection_id, action, status", - (node_type,), + summary["node_status"] = persistence_rows( + backend=backend, + db_path=target_db_path, + url=url, + sql=( + "SELECT collection_id, node_type, action, status, binding_status, COUNT(1) AS count " + f"FROM nodes WHERE node_type = '{escaped_node_type}' " + "GROUP BY collection_id, node_type, action, status, binding_status " + "ORDER BY collection_id, action, status" + ), ) return summary diff --git a/sync_state_machine/common/__init__.py b/sync_state_machine/common/__init__.py index bb690ad..a1cb34b 100644 --- a/sync_state_machine/common/__init__.py +++ b/sync_state_machine/common/__init__.py @@ -6,7 +6,7 @@ from .persistence import ( get_registered_persistence_backends, register_persistence_backend, ) -from .persistence_backends import SQLitePersistenceBackend +from .persistence_backends import SQLAlchemyPersistenceBackend, SQLitePersistenceBackend from .collection import DataCollection from .binding import BindingManager @@ -17,6 +17,7 @@ __all__ = [ "SyncNode", "PersistenceBackend", "SQLitePersistenceBackend", + "SQLAlchemyPersistenceBackend", "create_persistence_backend", "register_persistence_backend", "get_registered_persistence_backends", diff --git a/sync_state_machine/common/binding.py b/sync_state_machine/common/binding.py index ec633b9..36d3696 100644 --- a/sync_state_machine/common/binding.py +++ b/sync_state_machine/common/binding.py @@ -2,7 +2,7 @@ import json import logging from typing import Optional, Dict, Any, List -from ..logging import get_logger +from ..logging import get_logger, get_scope_display_name from .persistence import PersistenceBackend from .persistence_service import ScopedPersistenceView from .types import BindingStatus, BindingRecord @@ -253,7 +253,7 @@ class BindingManager: touched_text = ",".join(touched_types) if touched_types else "none" logger.info( - f"🔍 [BindingManager.persist] 完成: scope={self.scope}, types={type_count}, saved={saved_total}, deleted={deleted_total}, touched=[{touched_text}]" + f"🔍 [BindingManager.persist] 完成: scope={get_scope_display_name(self.scope)}, types={type_count}, saved={saved_total}, deleted={deleted_total}, touched=[{touched_text}]" ) async def unload(self, *, node_types: Optional[List[str]] = None) -> None: diff --git a/sync_state_machine/common/persistence.py b/sync_state_machine/common/persistence.py index 8dbee25..ae65c49 100644 --- a/sync_state_machine/common/persistence.py +++ b/sync_state_machine/common/persistence.py @@ -5,6 +5,9 @@ import sqlite3 from abc import ABC, abstractmethod from typing import Any, Callable, ClassVar, Dict, List, Optional, TYPE_CHECKING +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.engine import URL, make_url + from ..config.persist_config import PersistConfig if TYPE_CHECKING: @@ -26,72 +29,153 @@ class PersistenceBackend(ABC): cls, *, backend: str, - db_path: str, + db_path: str | None = None, + url: str | None = None, sql: str, limit: int = 200, ) -> tuple[List[str], List[Dict[str, Any]]]: - if backend != "sqlite": - raise NotImplementedError(f"records query backend not supported yet: {backend}") if not cls._is_read_only_sql(sql): raise ValueError("Only read-only SQL is allowed") - path = db_path - if not os.path.exists(path): - raise FileNotFoundError(f"DB not found: {path}") + if backend == "sqlite": + path = db_path + if not path: + raise ValueError("db_path is required for backend=sqlite") + if not os.path.exists(path): + raise FileNotFoundError(f"DB not found: {path}") - with sqlite3.connect(path) as conn: - conn.row_factory = sqlite3.Row - cursor = conn.execute(sql) - rows = cursor.fetchmany(max(1, min(limit, 2000))) - columns = [item[0] for item in (cursor.description or [])] - data = [{k: row[k] for k in columns} for row in rows] - return columns, data + with sqlite3.connect(path) as conn: + conn.row_factory = sqlite3.Row + cursor = conn.execute(sql) + rows = cursor.fetchmany(max(1, min(limit, 2000))) + columns = [item[0] for item in (cursor.description or [])] + data = [{k: row[k] for k in columns} for row in rows] + return columns, data + + if backend == "sqlalchemy": + if not url: + raise ValueError("url is required for backend=sqlalchemy") + engine = create_engine(cls._to_sync_sqlalchemy_url(url)) + try: + with engine.connect() as conn: + result = conn.execute(text(sql)) + rows = result.fetchmany(max(1, min(limit, 2000))) + columns = list(result.keys()) + data = [{column: row[index] for index, column in enumerate(columns)} for row in rows] + return columns, data + finally: + engine.dispose() + + raise NotImplementedError(f"records query backend not supported yet: {backend}") @classmethod def records_summary( cls, *, backend: str, - db_path: str, + db_path: str | None = None, + url: str | None = None, ) -> Dict[str, Any]: + db_exists = cls._target_exists(backend=backend, db_path=db_path, url=url) summary: Dict[str, Any] = { "backend": backend, "db_path": db_path, - "db_exists": os.path.exists(db_path), + "url": url, + "db_exists": db_exists, "tables": [], "sample_bindings": [], } - if not os.path.exists(db_path): - return summary - if backend != "sqlite": - summary["error"] = f"records summary backend not supported yet: {backend}" + if not db_exists: return summary - with sqlite3.connect(db_path) as conn: - conn.row_factory = sqlite3.Row - tables = conn.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name" - ).fetchall() - table_names = [str(r["name"]) for r in tables] - - for name in table_names: - count = conn.execute(f"SELECT COUNT(1) AS c FROM {name}").fetchone()["c"] - summary["tables"].append({"name": name, "count": int(count)}) - - if "bindings" in table_names: - sample = conn.execute( - "SELECT node_type, local_id, remote_id FROM bindings ORDER BY last_updated DESC LIMIT 50" + if backend == "sqlite": + assert db_path is not None + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + tables = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name" ).fetchall() - summary["sample_bindings"] = [ - { - "node_type": row["node_type"], - "local_id": row["local_id"], - "remote_id": row["remote_id"], - } - for row in sample - ] + table_names = [str(r["name"]) for r in tables] + + for name in table_names: + count = conn.execute(f"SELECT COUNT(1) AS c FROM {name}").fetchone()["c"] + summary["tables"].append({"name": name, "count": int(count)}) + + if "bindings" in table_names: + sample = conn.execute( + "SELECT node_type, local_id, remote_id FROM bindings ORDER BY last_updated DESC LIMIT 50" + ).fetchall() + summary["sample_bindings"] = [ + { + "node_type": row["node_type"], + "local_id": row["local_id"], + "remote_id": row["remote_id"], + } + for row in sample + ] + return summary + + if backend == "sqlalchemy": + if not url: + summary["error"] = "url is required for backend=sqlalchemy" + return summary + engine = create_engine(cls._to_sync_sqlalchemy_url(url)) + try: + with engine.connect() as conn: + inspector = inspect(conn) + table_names = sorted(inspector.get_table_names()) + + for name in table_names: + count = conn.execute(text(f"SELECT COUNT(1) AS c FROM {name}")).scalar_one() + summary["tables"].append({"name": name, "count": int(count)}) + + if "bindings" in table_names: + sample = conn.execute( + text("SELECT node_type, local_id, remote_id FROM bindings ORDER BY last_updated DESC LIMIT 50") + ).fetchall() + summary["sample_bindings"] = [ + { + "node_type": row[0], + "local_id": row[1], + "remote_id": row[2], + } + for row in sample + ] + finally: + engine.dispose() + return summary + + summary["error"] = f"records summary backend not supported yet: {backend}" return summary + @staticmethod + def _target_exists(*, backend: str, db_path: str | None, url: str | None) -> bool: + if backend == "sqlite": + return bool(db_path) and os.path.exists(db_path) + if backend == "sqlalchemy": + if not url: + return False + parsed = make_url(url) + if parsed.get_backend_name() == "sqlite": + database = parsed.database + if not database or database == ":memory:": + return True + return os.path.exists(database) + return True + return False + + @staticmethod + def _to_sync_sqlalchemy_url(url: str) -> str: + parsed = make_url(url) + async_to_sync = { + "sqlite+aiosqlite": "sqlite+pysqlite", + "postgresql+asyncpg": "postgresql+psycopg", + "mysql+aiomysql": "mysql+pymysql", + } + drivername = async_to_sync.get(parsed.drivername, parsed.drivername) + sync_url: URL = parsed.set(drivername=drivername) + return str(sync_url) + @abstractmethod async def initialize(self) -> None: raise NotImplementedError @@ -218,9 +302,11 @@ def create_persistence_backend( def _register_builtin_persistence_backends() -> None: + from .persistence_backends.sqlalchemy_backend import SQLAlchemyPersistenceBackend from .persistence_backends.sqlite_backend import SQLitePersistenceBackend register_persistence_backend("sqlite", lambda config: SQLitePersistenceBackend(config), override=True) + register_persistence_backend("sqlalchemy", lambda config: SQLAlchemyPersistenceBackend(config), override=True) _register_builtin_persistence_backends() diff --git a/sync_state_machine/common/persistence_backends/__init__.py b/sync_state_machine/common/persistence_backends/__init__.py index b20a62b..3571cdc 100644 --- a/sync_state_machine/common/persistence_backends/__init__.py +++ b/sync_state_machine/common/persistence_backends/__init__.py @@ -1,3 +1,4 @@ +from .sqlalchemy_backend import SQLAlchemyPersistenceBackend from .sqlite_backend import SQLitePersistenceBackend -__all__ = ["SQLitePersistenceBackend"] \ No newline at end of file +__all__ = ["SQLitePersistenceBackend", "SQLAlchemyPersistenceBackend"] \ No newline at end of file diff --git a/sync_state_machine/common/persistence_backends/sqlalchemy_backend.py b/sync_state_machine/common/persistence_backends/sqlalchemy_backend.py new file mode 100644 index 0000000..6a29068 --- /dev/null +++ b/sync_state_machine/common/persistence_backends/sqlalchemy_backend.py @@ -0,0 +1,428 @@ +from __future__ import annotations + +import json +import os +from typing import Any, Dict, List, Optional, TYPE_CHECKING + +from sqlalchemy import Column, Index, MetaData, PrimaryKeyConstraint, String, Table, Text, delete, insert, select +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncConnection, create_async_engine + +from ...config.persist_config import PersistConfig +from ..persistence import PersistenceBackend + +if TYPE_CHECKING: + from ..sync_node import SyncNode + + +class SQLAlchemyPersistenceBackend(PersistenceBackend): + """Generic SQLAlchemy persistence backend with explicit schema bootstrap.""" + + backend_name = "sqlalchemy" + bulk_delete_batch_size = 1000 + collection_id_length = 255 + scope_length = 255 + node_id_length = 255 + node_type_length = 128 + data_id_length = 255 + bind_data_id_length = 255 + action_length = 32 + status_length = 32 + binding_status_length = 32 + remote_id_length = 255 + last_updated_length = 64 + + def __init__(self, persist_config: PersistConfig | str | None = None, **_: Any) -> None: + if isinstance(persist_config, PersistConfig): + self.persist_config = persist_config + self.database_url = persist_config.resolved_url() + self.db_path = persist_config.resolved_db_path() + self.backend_options = dict(persist_config.backend_options) + elif isinstance(persist_config, str): + self.persist_config = None + self.database_url = persist_config + parsed = make_url(persist_config) + self.db_path = str(parsed.database) if parsed.get_backend_name() == "sqlite" and parsed.database else None + self.backend_options = {} + else: + raise ValueError("SQLAlchemyPersistenceBackend requires PersistConfig or database URL") + + self.engine: Optional[AsyncEngine] = None + self.metadata = MetaData() + self.nodes = Table( + "nodes", + self.metadata, + Column("collection_id", String(self.collection_id_length), nullable=False), + Column("scope", String(self.scope_length), nullable=False, default="global"), + Column("node_id", String(self.node_id_length), nullable=False), + Column("node_type", String(self.node_type_length), nullable=False), + Column("data_id", String(self.data_id_length), nullable=True), + Column("bind_data_id", String(self.bind_data_id_length), nullable=True), + Column("data", Text, nullable=True), + Column("origin_data", Text, nullable=True), + Column("action", String(self.action_length), nullable=True), + Column("status", String(self.status_length), nullable=True), + Column("binding_status", String(self.binding_status_length), nullable=True), + Column("error", Text, nullable=True), + Column("sync_log", Text, nullable=True), + Column("context", Text, nullable=True), + PrimaryKeyConstraint("collection_id", "scope", "node_id", name="pk_nodes"), + Index("idx_nodes_collection_scope_type", "collection_id", "scope", "node_type"), + Index("idx_nodes_collection_scope_data_id", "collection_id", "scope", "data_id"), + Index("idx_nodes_scope_type_node_id", "scope", "node_type", "node_id"), + ) + self.bindings = Table( + "bindings", + self.metadata, + Column("scope", String(self.scope_length), nullable=False, default="global"), + Column("node_type", String(self.node_type_length), nullable=False), + Column("local_id", String(self.node_id_length), nullable=False), + Column("remote_id", String(self.remote_id_length), nullable=True), + Column("last_updated", String(self.last_updated_length), nullable=True), + PrimaryKeyConstraint("scope", "node_type", "local_id", name="pk_bindings"), + Index("idx_bindings_scope_type_remote", "scope", "node_type", "remote_id"), + ) + + async def initialize(self) -> None: + if self.db_path and self.db_path != ":memory:": + dir_path = os.path.dirname(self.db_path) + if dir_path: + os.makedirs(dir_path, exist_ok=True) + + engine_options = {"future": True} + engine_options.update(self.backend_options) + self.engine = create_async_engine(self.database_url, **engine_options) + async with self.engine.begin() as conn: + await conn.run_sync(self.metadata.create_all) + + async def save_node(self, collection_id: str, scope: str, node: "SyncNode") -> None: + row = self._node_row(collection_id=collection_id, scope=scope, node=node) + async with self._begin() as conn: + await self._replace_node_rows(conn, collection_id=collection_id, scope=scope, node_ids=[node.node_id]) + await conn.execute(insert(self.nodes), [row]) + + async def save_nodes_bulk(self, collection_id: str, scope: str, nodes: List["SyncNode"]) -> None: + if not nodes: + return + rows = [self._node_row(collection_id=collection_id, scope=scope, node=node) for node in nodes] + node_ids = [node.node_id for node in nodes] + async with self._begin() as conn: + await self._replace_node_rows(conn, collection_id=collection_id, scope=scope, node_ids=node_ids) + await conn.execute(insert(self.nodes), rows) + + async def delete_node(self, collection_id: str, scope: str, node_id: str) -> None: + async with self._begin() as conn: + await conn.execute( + delete(self.nodes).where( + self.nodes.c.collection_id == collection_id, + self.nodes.c.scope == scope, + self.nodes.c.node_id == node_id, + ) + ) + + async def delete_nodes_bulk(self, collection_id: str, scope: str, node_ids: List[str]) -> None: + if not node_ids: + return + async with self._begin() as conn: + for batch in self._chunked(node_ids): + await conn.execute( + delete(self.nodes).where( + self.nodes.c.collection_id == collection_id, + self.nodes.c.scope == scope, + self.nodes.c.node_id.in_(batch), + ) + ) + + async def load_nodes( + self, + collection_id: str, + scope: str, + *, + exclude_node_types: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: + stmt = select(self.nodes).where( + self.nodes.c.collection_id == collection_id, + self.nodes.c.scope == scope, + ) + filtered_types = [node_type for node_type in (exclude_node_types or []) if node_type] + if filtered_types: + stmt = stmt.where(self.nodes.c.node_type.not_in(filtered_types)) + + async with self._connect() as conn: + rows = (await conn.execute(stmt)).mappings().all() + return [self._deserialize_node_row(row) for row in rows] + + async def save_binding(self, scope: str, node_type: str, local_id: str, payload_dict: Dict[str, Any]) -> None: + row = self._binding_row(scope=scope, node_type=node_type, local_id=local_id, remote_id=payload_dict.get("remote_id") if payload_dict else None) + async with self._begin() as conn: + await self._replace_binding_rows(conn, scope=scope, node_type=node_type, local_ids=[local_id]) + await conn.execute(insert(self.bindings), [row]) + + async def save_bindings_bulk(self, scope: str, node_type: str, bindings: Dict[str, Optional[str]]) -> None: + if not bindings: + return + rows = [ + self._binding_row(scope=scope, node_type=node_type, local_id=local_id, remote_id=remote_id) + for local_id, remote_id in bindings.items() + ] + async with self._begin() as conn: + await self._replace_binding_rows(conn, scope=scope, node_type=node_type, local_ids=list(bindings.keys())) + await conn.execute(insert(self.bindings), rows) + + async def delete_binding(self, scope: str, node_type: str, local_id: str) -> None: + async with self._begin() as conn: + await conn.execute( + delete(self.bindings).where( + self.bindings.c.scope == scope, + self.bindings.c.node_type == node_type, + self.bindings.c.local_id == local_id, + ) + ) + + async def delete_bindings_bulk(self, scope: str, node_type: str, local_ids: List[str]) -> None: + if not local_ids: + return + async with self._begin() as conn: + for batch in self._chunked(local_ids): + await conn.execute( + delete(self.bindings).where( + self.bindings.c.scope == scope, + self.bindings.c.node_type == node_type, + self.bindings.c.local_id.in_(batch), + ) + ) + + async def load_bindings(self, scope: str, node_type: str) -> List[Dict[str, Any]]: + stmt = select(self.bindings.c.local_id, self.bindings.c.remote_id).where( + self.bindings.c.scope == scope, + self.bindings.c.node_type == node_type, + ) + async with self._connect() as conn: + rows = (await conn.execute(stmt)).mappings().all() + return [{"local_id": row["local_id"], "payload": {"remote_id": row["remote_id"]}} for row in rows] + + async def load_bind_data_id_overrides(self, node_type: str, scope: str) -> Dict[str, Optional[str]]: + local_join = ( + select( + self.bindings.c.local_id.label("src_node_id"), + self.nodes.c.data_id.label("bind_data_id"), + ) + .select_from( + self.bindings.outerjoin( + self.nodes, + (self.nodes.c.node_id == self.bindings.c.remote_id) + & (self.nodes.c.node_type == self.bindings.c.node_type) + & (self.nodes.c.scope == self.bindings.c.scope), + ) + ) + .where( + self.bindings.c.node_type == node_type, + self.bindings.c.scope == scope, + ) + ) + remote_join = ( + select( + self.bindings.c.remote_id.label("src_node_id"), + self.nodes.c.data_id.label("bind_data_id"), + ) + .select_from( + self.bindings.outerjoin( + self.nodes, + (self.nodes.c.node_id == self.bindings.c.local_id) + & (self.nodes.c.node_type == self.bindings.c.node_type) + & (self.nodes.c.scope == self.bindings.c.scope), + ) + ) + .where( + self.bindings.c.node_type == node_type, + self.bindings.c.scope == scope, + self.bindings.c.remote_id.is_not(None), + ) + ) + async with self._connect() as conn: + rows = (await conn.execute(local_join.union_all(remote_join))).mappings().all() + result: Dict[str, Optional[str]] = {} + for row in rows: + src_node_id = row["src_node_id"] + if src_node_id: + result[str(src_node_id)] = row["bind_data_id"] + return result + + async def delete_scope_node_types(self, *, scope: str, node_types: List[str]) -> None: + filtered_types = [node_type for node_type in node_types if node_type] + if not filtered_types: + return + async with self._begin() as conn: + await conn.execute(delete(self.nodes).where(self.nodes.c.scope == scope, self.nodes.c.node_type.in_(filtered_types))) + await conn.execute(delete(self.bindings).where(self.bindings.c.scope == scope, self.bindings.c.node_type.in_(filtered_types))) + + async def copy_nodes_between_scopes( + self, + *, + collection_id: str, + source_scope: str, + target_scope: str, + node_types: List[str], + ) -> None: + if source_scope == target_scope or not node_types: + return + stmt = select(self.nodes).where( + self.nodes.c.collection_id == collection_id, + self.nodes.c.scope == source_scope, + self.nodes.c.node_type.in_(node_types), + ) + async with self._begin() as conn: + rows = (await conn.execute(stmt)).mappings().all() + target_rows = [dict(row, scope=target_scope) for row in rows] + if not target_rows: + return + await self._replace_node_rows( + conn, + collection_id=collection_id, + scope=target_scope, + node_ids=[str(row["node_id"]) for row in target_rows], + ) + await conn.execute(insert(self.nodes), target_rows) + + async def copy_bindings_between_scopes( + self, + *, + source_scope: str, + target_scope: str, + node_types: List[str], + ) -> None: + if source_scope == target_scope or not node_types: + return + stmt = select(self.bindings).where( + self.bindings.c.scope == source_scope, + self.bindings.c.node_type.in_(node_types), + ) + async with self._begin() as conn: + rows = (await conn.execute(stmt)).mappings().all() + target_rows = [dict(row, scope=target_scope) for row in rows] + if not target_rows: + return + by_type: Dict[str, List[str]] = {} + for row in target_rows: + by_type.setdefault(str(row["node_type"]), []).append(str(row["local_id"])) + for node_type, local_ids in by_type.items(): + await self._replace_binding_rows(conn, scope=target_scope, node_type=node_type, local_ids=local_ids) + await conn.execute(insert(self.bindings), target_rows) + + async def dump_to_runtime(self, filename: str) -> None: + raise NotImplementedError("dump_to_runtime is only supported by the sqlite backend") + + async def close(self) -> None: + if self.engine is None: + return + await self.engine.dispose() + self.engine = None + + async def _replace_node_rows( + self, + conn: AsyncConnection, + *, + collection_id: str, + scope: str, + node_ids: List[str], + ) -> None: + if not node_ids: + return + for batch in self._chunked(node_ids): + await conn.execute( + delete(self.nodes).where( + self.nodes.c.collection_id == collection_id, + self.nodes.c.scope == scope, + self.nodes.c.node_id.in_(batch), + ) + ) + + async def _replace_binding_rows( + self, + conn: AsyncConnection, + *, + scope: str, + node_type: str, + local_ids: List[str], + ) -> None: + if not local_ids: + return + for batch in self._chunked(local_ids): + await conn.execute( + delete(self.bindings).where( + self.bindings.c.scope == scope, + self.bindings.c.node_type == node_type, + self.bindings.c.local_id.in_(batch), + ) + ) + + def _chunked(self, values: List[str]) -> List[List[str]]: + if not values: + return [] + batch_size = max(1, int(self.bulk_delete_batch_size)) + return [values[index : index + batch_size] for index in range(0, len(values), batch_size)] + + def _node_row(self, *, collection_id: str, scope: str, node: "SyncNode") -> Dict[str, Any]: + bind_data_id = None + if isinstance(node.context, dict): + value = node.context.get("bind_data_id") + if value is not None and value != "": + bind_data_id = str(value) + return { + "collection_id": collection_id, + "scope": scope, + "node_id": node.node_id, + "node_type": node.node_type, + "data_id": node.data_id, + "bind_data_id": bind_data_id, + "data": json.dumps(node.data) if node.data else None, + "origin_data": json.dumps(node.origin_data) if node.origin_data else None, + "action": node.action.value, + "status": node.status.value, + "binding_status": node.binding_status.value, + "error": node.error, + "sync_log": node.sync_log, + "context": json.dumps(node.context) if node.context else None, + } + + @staticmethod + def _binding_row(*, scope: str, node_type: str, local_id: str, remote_id: Optional[str]) -> Dict[str, Any]: + return { + "scope": scope, + "node_type": node_type, + "local_id": local_id, + "remote_id": remote_id, + "last_updated": None, + } + + @staticmethod + def _deserialize_node_row(row: Dict[str, Any]) -> Dict[str, Any]: + item = { + "node_id": row["node_id"], + "node_type": row["node_type"], + "data_id": row["data_id"], + "bind_data_id": row["bind_data_id"], + "depend_ids": [], + "data": json.loads(row["data"]) if row["data"] else None, + "origin_data": json.loads(row["origin_data"]) if row["origin_data"] else None, + "action": row["action"], + "status": row["status"], + "binding_status": row["binding_status"], + "error": row["error"], + "sync_log": row["sync_log"], + "context": json.loads(row["context"]) if row["context"] else {}, + } + if item["bind_data_id"] and isinstance(item["context"], dict): + item["context"]["bind_data_id"] = item["bind_data_id"] + return item + + def _require_engine(self) -> AsyncEngine: + if self.engine is None: + raise RuntimeError("SQLAlchemyPersistenceBackend is not initialized") + return self.engine + + def _connect(self): + return self._require_engine().connect() + + def _begin(self): + return self._require_engine().begin() \ No newline at end of file diff --git a/sync_state_machine/common/persistence_backends/sqlite_backend.py b/sync_state_machine/common/persistence_backends/sqlite_backend.py index ea4551b..38432f4 100644 --- a/sync_state_machine/common/persistence_backends/sqlite_backend.py +++ b/sync_state_machine/common/persistence_backends/sqlite_backend.py @@ -20,7 +20,10 @@ class SQLitePersistenceBackend(PersistenceBackend): def __init__(self, persist_config: PersistConfig | str | None = None, **_: Any) -> None: if isinstance(persist_config, PersistConfig): self.persist_config = persist_config - self.db_path = persist_config.db_path + resolved_db_path = persist_config.resolved_db_path() + if not resolved_db_path: + raise ValueError("sqlite persistence requires a resolvable db_path") + self.db_path = resolved_db_path elif isinstance(persist_config, str): self.persist_config = None self.db_path = persist_config diff --git a/sync_state_machine/common/sqlalchemy_backend.py b/sync_state_machine/common/sqlalchemy_backend.py new file mode 100644 index 0000000..d419dd9 --- /dev/null +++ b/sync_state_machine/common/sqlalchemy_backend.py @@ -0,0 +1,3 @@ +from .persistence_backends.sqlalchemy_backend import SQLAlchemyPersistenceBackend + +__all__ = ["SQLAlchemyPersistenceBackend"] \ No newline at end of file diff --git a/sync_state_machine/config/persist_config.py b/sync_state_machine/config/persist_config.py index 0d638d4..fcd568d 100644 --- a/sync_state_machine/config/persist_config.py +++ b/sync_state_machine/config/persist_config.py @@ -1,13 +1,70 @@ -from typing import Any, Dict +from typing import Any, Dict, Optional -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator class PersistConfig(BaseModel): model_config = ConfigDict(extra="forbid", validate_assignment=True) backend: str = "sqlite" - db_path: str + db_path: Optional[str] = None + url: Optional[str] = None backend_options: Dict[str, Any] = Field(default_factory=dict) enable: bool = True wipe_on_start: bool = False + + @model_validator(mode="before") + @classmethod + def _validate_target(cls, data: Any) -> Any: + if not isinstance(data, dict): + return data + + normalized = dict(data) + normalized["backend"] = (str(normalized.get("backend") or "sqlite").strip().lower() or "sqlite") + normalized["db_path"] = cls._normalize_optional_text(normalized.get("db_path")) + normalized["url"] = cls._normalize_optional_text(normalized.get("url")) + + if not normalized["db_path"] and not normalized["url"]: + raise ValueError("persist requires either db_path or url") + if normalized["backend"] != "sqlite" and not normalized["url"]: + raise ValueError(f"persist.url is required for backend={normalized['backend']}") + return normalized + + @staticmethod + def _normalize_optional_text(value: Optional[str]) -> Optional[str]: + if value is None: + return None + normalized = str(value).strip() + return normalized or None + + def resolved_url(self) -> str: + if self.url: + return self.url + if self.backend == "sqlite" and self.db_path: + return f"sqlite+aiosqlite:///{self.db_path}" + raise ValueError(f"persist.url is required for backend={self.backend}") + + def resolved_db_path(self) -> Optional[str]: + if self.db_path: + return self.db_path + if not self.url: + return None + + from sqlalchemy.engine import make_url + + parsed = make_url(self.url) + if parsed.get_backend_name() != "sqlite": + return None + database = parsed.database + if not database: + return None + return str(database) + + def wipe_file_path(self) -> Optional[str]: + db_path = self.resolved_db_path() + if not db_path or db_path == ":memory:": + return None + return db_path + + def display_target(self) -> str: + return self.url or self.db_path or "" diff --git a/sync_state_machine/datasource/api/client.py b/sync_state_machine/datasource/api/client.py index 9281389..2aa7345 100644 --- a/sync_state_machine/datasource/api/client.py +++ b/sync_state_machine/datasource/api/client.py @@ -23,7 +23,7 @@ from datetime import datetime import httpx from schemas.common.push_system import PushIdsSchema -from ...logging import get_logger +from ...logging import get_logger, get_scope_display_name def _build_default_push_steps() -> List[Dict[str, str]]: @@ -43,6 +43,7 @@ class ApiClient: secret: str, debug: bool = False, log_file: Optional[Path] = None, + session_label: Optional[str] = None, rate_limit_max_requests: int = 30, rate_limit_window_seconds: float = 10.0, ): @@ -64,6 +65,8 @@ class ApiClient: self._client: Optional[httpx.AsyncClient] = None self._debug = debug self._log_file = log_file + self._scope = "" + self._session_label = (session_label or "").strip() self._trace_seq = 0 self._trace_by_push_id: Dict[str, deque[Dict[str, Any]]] = defaultdict(deque) self._trace_by_biz_id: Dict[str, deque[Dict[str, Any]]] = defaultdict(deque) @@ -78,6 +81,43 @@ class ApiClient: self._request_success = 0 self._request_failure = 0 + def set_scope(self, scope: Optional[str]) -> None: + self._scope = str(scope or "").strip() + + def set_session_label(self, label: Optional[str]) -> None: + self._session_label = str(label or "").strip() + + def get_log_session_label(self) -> str: + if self._session_label: + return self._session_label + scope_label = get_scope_display_name(self._scope) + return scope_label or self._scope + + def _project_name_from_request(self, payload: Optional[Dict[str, Any]]) -> str: + if not isinstance(payload, dict): + return "" + project_id = str(payload.get("project_id") or "").strip() + if not project_id: + return "" + display = get_scope_display_name(f"project_id:{project_id}") + if not display or display == f"project_id:{project_id}": + return "" + return display + + def _resolve_request_session_label( + self, + *, + request_params: Optional[Dict[str, Any]] = None, + request_payload: Optional[Dict[str, Any]] = None, + ) -> str: + payload_label = self._project_name_from_request(request_payload) + if payload_label: + return payload_label + params_label = self._project_name_from_request(request_params) + if params_label: + return params_label + return self.get_log_session_label() + def _is_rate_limit_enabled(self) -> bool: return self._rate_limit_max_requests > 0 and self._rate_limit_window_seconds > 0 @@ -110,7 +150,7 @@ class ApiClient: async def close(self): """关闭 HTTP 客户端""" - self.log_request_stats() + self.log_request_stats(level="INFO") if self._client: await self._client.aclose() self._client = None @@ -126,8 +166,11 @@ class ApiClient: def format_request_stats(self) -> str: """格式化当前客户端生命周期内的请求统计。""" stats = self.get_request_stats() + session_text = self.get_log_session_label() + session_prefix = f"session={session_text} " if session_text else "" return ( f"{self._REQUEST_STATS_LOG_PREFIX}: " + f"{session_prefix}" f"total={stats['total']}, success={stats['success']}, failure={stats['failure']}" ) @@ -135,12 +178,25 @@ class ApiClient: """输出当前客户端生命周期内的请求统计。""" self._log(self.format_request_stats(), level=level) - def _log_request_summary(self, *, method: str, endpoint: str, elapsed_seconds: float) -> None: + def _log_request_summary( + self, + *, + method: str, + endpoint: str, + elapsed_seconds: float, + request_params: Optional[Dict[str, Any]] = None, + request_payload: Optional[Dict[str, Any]] = None, + ) -> None: if not self._debug: return + session_text = self._resolve_request_session_label( + request_params=request_params, + request_payload=request_payload, + ) + session_prefix = f"session={session_text} " if session_text else "" self._log( - f"🔎 ApiClient request: method={method.upper()} endpoint={endpoint} elapsed={elapsed_seconds:.3f}s", - level="DEBUG", + f"🔎 ApiClient request: {session_prefix}method={method.upper()} endpoint={endpoint} elapsed={elapsed_seconds:.3f}s", + level="INFO", ) def _log(self, message: str, level: str = "DEBUG"): @@ -468,7 +524,13 @@ class ApiClient: # 记录响应时间 elapsed = time.time() - start_time - self._log_request_summary(method=method, endpoint=endpoint, elapsed_seconds=elapsed) + self._log_request_summary( + method=method, + endpoint=endpoint, + elapsed_seconds=elapsed, + request_params=params, + request_payload=data if method.upper() in ['POST', 'PUT', 'DELETE'] else None, + ) self._log(f"Response Time: {elapsed:.3f}s", level="DEBUG") # 检查 HTTP 状态 @@ -499,7 +561,13 @@ class ApiClient: # 记录错误 elapsed = time.time() - start_time - self._log_request_summary(method=method, endpoint=endpoint, elapsed_seconds=elapsed) + self._log_request_summary( + method=method, + endpoint=endpoint, + elapsed_seconds=elapsed, + request_params=params, + request_payload=data if method.upper() in ['POST', 'PUT', 'DELETE'] else None, + ) error_msg = "\n========== [ApiClient HTTP Error] ==========\n" error_msg += f"Request: {method} {url}\n" error_msg += f"Params: {params}\n" diff --git a/sync_state_machine/datasource/api/datasource.py b/sync_state_machine/datasource/api/datasource.py index b68e0cc..ff95722 100644 --- a/sync_state_machine/datasource/api/datasource.py +++ b/sync_state_machine/datasource/api/datasource.py @@ -15,6 +15,7 @@ from ..datasource import BaseDataSource from ..handler import NodeHandler from ..task_result import TaskResult from .client import ApiClient +from ...logging import set_scope_display_name class ApiDataSource(BaseDataSource): @@ -56,6 +57,29 @@ class ApiDataSource(BaseDataSource): rate_limit_window_seconds=rate_limit_window_seconds, ) self.poll_mode = poll_mode + + def set_collection(self, collection) -> None: + super().set_collection(collection) + self.client.set_scope(getattr(collection, "scope", "")) + + def _update_client_session_label_from_project_node(self, node) -> None: + if self._collection is None or getattr(self._collection, "scope", "") == "global": + return + if getattr(node, "node_type", None) != "project": + return + if not hasattr(node, "get_data"): + return + + data = node.get_data() + if not isinstance(data, dict): + return + + project_name = str(data.get("name") or "").strip() + if not project_name: + return + + set_scope_display_name(self._collection.scope, f"project:{project_name}") + self.client.set_session_label(f"project:{project_name}") async def initialize(self): """初始化数据源（初始化 HTTP 客户端）""" @@ -175,6 +199,7 @@ class ApiDataSource(BaseDataSource): return None def _on_node_loaded(self, handler: NodeHandler, node) -> None: + self._update_client_session_label_from_project_node(node) data_id = node.data_id or "" if not data_id: return diff --git a/sync_state_machine/datasource/datasource.py b/sync_state_machine/datasource/datasource.py index c3c79ff..71b28bc 100644 --- a/sync_state_machine/datasource/datasource.py +++ b/sync_state_machine/datasource/datasource.py @@ -37,7 +37,7 @@ from ..engine import ( StateMachineRuntime, e40_sync_execute, ) -from ..logging import get_logger +from ..logging import get_logger, get_scope_display_name logger = get_logger(__name__) @@ -95,7 +95,7 @@ class BaseDataSource(ABC): def _datasource_log_prefix(self) -> str: scope = self._collection.scope if self._collection is not None else "unbound" - return f"[{scope}][{self._endpoint_name}:{self._datasource_type}]" + return f"[{get_scope_display_name(scope)}][{self._endpoint_name}:{self._datasource_type}]" def _node_log_prefix(self, node_type: str) -> str: return f"{self._datasource_log_prefix()}[{node_type}]" diff --git a/sync_state_machine/logging.py b/sync_state_machine/logging.py index 924452a..8b8ee30 100644 --- a/sync_state_machine/logging.py +++ b/sync_state_machine/logging.py @@ -6,11 +6,34 @@ from typing import Optional APP_LOGGER_NAME = "sync_state_machine" +_SCOPE_DISPLAY_NAMES: dict[str, str] = {} + + +def set_scope_display_name(scope: str, display_name: str) -> None: + normalized_scope = str(scope).strip() + normalized_name = str(display_name).strip() + if not normalized_scope: + return + if not normalized_name: + _SCOPE_DISPLAY_NAMES.pop(normalized_scope, None) + return + _SCOPE_DISPLAY_NAMES[normalized_scope] = normalized_name + + +def get_scope_display_name(scope: Optional[str]) -> str: + normalized_scope = str(scope or "").strip() + if not normalized_scope: + return "" + return _SCOPE_DISPLAY_NAMES.get(normalized_scope, normalized_scope) + + +def clear_scope_display_names() -> None: + _SCOPE_DISPLAY_NAMES.clear() class ScopeLoggerAdapter(logging.LoggerAdapter): def process(self, msg, kwargs): - scope = self.extra.get("scope") + scope = get_scope_display_name(self.extra.get("scope")) if scope and not str(msg).startswith("["): msg = f"[{scope}] {msg}" return msg, kwargs diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index a874388..3a80a7e 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -197,7 +197,7 @@ def _log_pipeline_start(config: PipelineRunConfig, *, title: str) -> None: def _log_pipeline_end(config: PipelineRunConfig, *, stats: Dict[str, Any]) -> None: logger.info("✅ Pipeline completed: stats=%s", stats) - logger.info("🗃️ Persistence DB: %s", config.persist.db_path) + logger.info("🗃️ Persistence Target: %s", config.persist.display_target()) ensure_builtin_datasource_types_registered() @@ -428,15 +428,17 @@ async def create_scope_runtime_from_config( ) -> ProjectScopeRuntime: ensure_app_logging(config.logging) - db_path = Path(config.persist.db_path) owns_persistence_service = persistence_service is None created_backend: Optional[PersistenceBackend] = None local_ds = None remote_ds = None try: if persistence_service is None: - if config.persist.wipe_on_start and db_path.exists(): - db_path.unlink() + wipe_file_path = config.persist.wipe_file_path() + if config.persist.wipe_on_start and wipe_file_path: + db_path = Path(wipe_file_path) + if db_path.exists(): + db_path.unlink() created_backend = create_persistence_backend(config.persist) persistence_service = PersistenceService(created_backend) await persistence_service.initialize() diff --git a/sync_state_machine/pipeline/multi_project_pipeline.py b/sync_state_machine/pipeline/multi_project_pipeline.py index c4269d7..c458a40 100644 --- a/sync_state_machine/pipeline/multi_project_pipeline.py +++ b/sync_state_machine/pipeline/multi_project_pipeline.py @@ -14,7 +14,7 @@ from ..config import ( apply_config_overrides, build_config_from_file, ) -from ..logging import get_scope_logger +from ..logging import get_scope_logger, get_scope_display_name from .factory import create_pipeline_from_config, create_scope_runtime_from_config @@ -337,7 +337,7 @@ class ProjectBatchPipeline: title = f"multi-project pipeline [{scope}]" logger.info( "🚀 Project pipeline start: scope=%s local_project_id=%s remote_project_id=%s", - scope, + get_scope_display_name(scope), item.local_project_id, item.remote_project_id, ) diff --git a/sync_state_machine/pipeline/summary_report.py b/sync_state_machine/pipeline/summary_report.py index 6745eb6..5eeeca2 100644 --- a/sync_state_machine/pipeline/summary_report.py +++ b/sync_state_machine/pipeline/summary_report.py @@ -1,7 +1,18 @@ from __future__ import annotations +import logging + async def print_pipeline_summary(*, logger, node_types, binding_manager, local_datasource, remote_datasource, local_collection, remote_collection, consistency_by_type, stats) -> None: + scoped_logger = logger + plain_logger = logger.logger if isinstance(logger, logging.LoggerAdapter) else logger + + def _log_header(message: str) -> None: + scoped_logger.info(message) + + def _log_body(message: str) -> None: + plain_logger.info(message) + binding_counts = {} for node_type in node_types: records = await binding_manager.get_all_records(node_type) @@ -36,31 +47,31 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d remote_text = _clip(repr(sample.get("remote")), 72) return f"{local_text}/{remote_text}" - logger.info(f"\n{'=' * line_width}") - logger.info("🔎 Consistency Summary") - logger.info(f"{'=' * line_width}") + _log_header(f"\n{'=' * line_width}") + _log_header("🔎 Consistency Summary") + _log_header(f"{'=' * line_width}") for node_type in node_types: item = consistency_by_type.get(node_type, {}) or {} checked_pairs = int(item.get("checked_pairs", 0)) mismatch_count = int(item.get("mismatch_count", 0)) - logger.info(f"{node_type} (checked_pairs={checked_pairs}, mismatches={mismatch_count})") + _log_body(f"{node_type} (checked_pairs={checked_pairs}, mismatches={mismatch_count})") fields = item.get("fields", []) or [] differing_fields = [field for field in fields if int(field.get("mismatch_count", 0)) > 0] if not differing_fields: - logger.info(" pass") + _log_body(" pass") continue - logger.info(" field mismatch/total samples") - logger.info(" -----------------------------------------------------------") + _log_body(" field mismatch/total samples") + _log_body(" -----------------------------------------------------------") 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)) samples = field_report.get("samples", []) or [] sample_text = _format_sample(samples[0]) if samples else "-" - logger.info( + _log_body( f" {_clip(field_name, 30):<30} {field_mismatch:>4}/{field_total:<4} " f"{sample_text}" ) @@ -74,15 +85,15 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d action_w = 16 consistency_w = 18 - logger.info(f"\n{'='*line_width}") - logger.info(f"📊 {title} Summary") - logger.info(f"{'='*line_width}") - logger.info("说明: Create/Update/Delete = 成功/失败/总数 (失败=FAILED+SKIPPED)") - logger.info("说明: Consistent = 一致对数/不一致对数/已检查绑定对数") - logger.info( + _log_header(f"\n{'='*line_width}") + _log_header(f"📊 {title} Summary") + _log_header(f"{'='*line_width}") + _log_body("说明: Create/Update/Delete = 成功/失败/总数 (失败=FAILED+SKIPPED)") + _log_body("说明: Consistent = 一致对数/不一致对数/已检查绑定对数") + _log_body( f"{'Node Type':<{node_type_w}} {'Bound/Rec/Load':<{bound_w}} {'Create(S/F/T)':<{action_w}} {'Update(S/F/T)':<{action_w}} {'Delete(S/F/T)':<{action_w}} {'Consistent(S/F/T)':<{consistency_w}}" ) - logger.info(f"{'-'*line_width}") + _log_body(f"{'-'*line_width}") total_bound_normal = 0 total_bindings = 0 @@ -105,7 +116,7 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d consistency_str = format_consistency_sft(node_type) bound_str = f"{bound_normal_count}/{bindings}/{loaded_count}" - logger.info( + _log_body( f"{_clip(node_type, node_type_w):<{node_type_w}} " f"{_clip(bound_str, bound_w):<{bound_w}} " f"{_clip(create_str, action_w):<{action_w}} " @@ -131,14 +142,14 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d total_consistency["checked"] += int(consistency.get("checked_pairs", 0)) total_consistency["mismatch"] += int(consistency.get("mismatch_count", 0)) - logger.info(f"{'-'*line_width}") + _log_body(f"{'-'*line_width}") total_consistent = max(0, total_consistency["checked"] - total_consistency["mismatch"]) total_create_sft = f"{total_create['success']}/{total_create['failed_effective']}/{total_create['total']}" total_update_sft = f"{total_update['success']}/{total_update['failed_effective']}/{total_update['total']}" total_delete_sft = f"{total_delete['success']}/{total_delete['failed_effective']}/{total_delete['total']}" total_consistency_sft = f"{total_consistent}/{total_consistency['mismatch']}/{total_consistency['checked']}" total_bound_str = f"{total_bound_normal}/{total_bindings}/{total_loaded}" - logger.info( + _log_body( f"{'TOTAL':<{node_type_w}} " f"{_clip(total_bound_str, bound_w):<{bound_w}} " f"{_clip(total_create_sft, action_w):<{action_w}} " @@ -158,13 +169,13 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d failed = int(stats.get("failed", 0)) post_check_passed = bool(stats.get("post_check_passed", True)) - logger.info("\n" + "=" * 96) - logger.info( + _log_header("\n" + "=" * 96) + _log_header( f"📈 Pipeline Stats: loaded={loaded}, synced={synced}, failed={failed}, post_check_passed={post_check_passed}, mismatch_types={len(mismatch_node_types)}" ) if mismatch_node_types: - logger.info(f"🔎 Mismatch Node Types: {', '.join(mismatch_node_types)}") - logger.info("=" * 96) + _log_body(f"🔎 Mismatch Node Types: {', '.join(mismatch_node_types)}") + _log_header("=" * 96) print_consistency_summary() print_collection_summary("Local", local_datasource, local_collection) diff --git a/tests/unit/test_api_client_rate_limit.py b/tests/unit/test_api_client_rate_limit.py index bd3dd31..15dd5c0 100644 --- a/tests/unit/test_api_client_rate_limit.py +++ b/tests/unit/test_api_client_rate_limit.py @@ -10,6 +10,8 @@ import pytest import httpx from sync_state_machine.datasource.api.client import ApiClient +from sync_state_machine.datasource.api.datasource import ApiDataSource +from sync_state_machine.logging import clear_scope_display_names, set_scope_display_name class _FakeResponse: @@ -50,6 +52,25 @@ class _FailingAsyncClient: self.closed = True +class _FakeCollection: + def __init__(self, scope: str) -> None: + self.scope = scope + + def set_state_machine_runtime(self, runtime) -> None: + del runtime + + +class _FakeProjectNode: + node_type = "project" + + def __init__(self, data_id: str, name: str) -> None: + self.data_id = data_id + self._data = {"name": name} + + def get_data(self) -> dict: + return dict(self._data) + + @pytest.mark.asyncio async def test_api_client_global_rate_limit_applies_to_concurrent_requests() -> None: client = ApiClient( @@ -120,6 +141,78 @@ async def test_api_client_logs_one_line_request_summary_in_debug_mode(caplog: py assert "🔎 ApiClient request: method=GET endpoint=/ping elapsed=" in caplog.text +@pytest.mark.asyncio +async def test_api_client_request_summary_includes_session_label_with_scope_fallback( + caplog: pytest.LogCaptureFixture, +) -> None: + clear_scope_display_names() + client = ApiClient( + base_url="http://example.com", + uid="uid", + secret="secret", + debug=True, + ) + logger_name = "sync_state_machine.datasource.api.client" + fake_client = _FakeAsyncClient() + client._client = fake_client + client.set_scope("project_id:project-1") + + with caplog.at_level(logging.INFO, logger=logger_name): + await client.request("GET", "/ping") + + assert "session=project_id:project-1 method=GET endpoint=/ping" in caplog.text + + caplog.clear() + client.set_session_label("project:Alpha") + + with caplog.at_level(logging.INFO, logger=logger_name): + await client.request("GET", "/ping") + + assert "session=project:Alpha method=GET endpoint=/ping" in caplog.text + clear_scope_display_names() + + +@pytest.mark.asyncio +async def test_api_client_request_summary_uses_registered_project_name_for_project_id_requests( + caplog: pytest.LogCaptureFixture, +) -> None: + clear_scope_display_names() + set_scope_display_name("project_id:project-1", "project:Alpha") + client = ApiClient( + base_url="http://example.com", + uid="uid", + secret="secret", + debug=True, + ) + logger_name = "sync_state_machine.datasource.api.client" + fake_client = _FakeAsyncClient() + client._client = fake_client + client.set_scope("global") + + with caplog.at_level(logging.INFO, logger=logger_name): + await client.request("GET", "/supplier/list", params={"project_id": "project-1"}) + + clear_scope_display_names() + assert "session=project:Alpha method=GET endpoint=/supplier/list" in caplog.text + + +def test_api_datasource_updates_client_session_label_from_project_node() -> None: + clear_scope_display_names() + datasource = ApiDataSource( + base_url="http://example.com", + uid="uid", + secret="secret", + debug=True, + ) + + datasource.set_collection(_FakeCollection("project_id:project-1")) + assert datasource.client.get_log_session_label() == "project_id:project-1" + + datasource._on_node_loaded(None, _FakeProjectNode(data_id="project-1", name="Alpha")) + assert datasource.client.get_log_session_label() == "project:Alpha" + clear_scope_display_names() + + @pytest.mark.asyncio async def test_api_client_get_signature_keeps_list_params_in_signature_source() -> None: client = ApiClient( diff --git a/tests/unit/test_persist_config_and_sqlalchemy_backend.py b/tests/unit/test_persist_config_and_sqlalchemy_backend.py new file mode 100644 index 0000000..c6b5a85 --- /dev/null +++ b/tests/unit/test_persist_config_and_sqlalchemy_backend.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest +from sqlalchemy.dialects import mysql +from sqlalchemy.schema import CreateTable + +from sync_state_machine.common.binding import BindingManager +from sync_state_machine.common.collection import DataCollection +from sync_state_machine.common.persistence import PersistenceBackend, create_persistence_backend +from sync_state_machine.common.sqlalchemy_backend import SQLAlchemyPersistenceBackend +from sync_state_machine.config.persist_config import PersistConfig +from tests.fixtures.mock_domain import build_project_node, register_test_domain + + +def test_persist_config_accepts_sqlalchemy_url_only() -> None: + cfg = PersistConfig( + backend="sqlalchemy", + url="postgresql+asyncpg://user:pass@localhost:5432/persist_db", + ) + + assert cfg.resolved_url() == "postgresql+asyncpg://user:pass@localhost:5432/persist_db" + assert cfg.resolved_db_path() is None + assert cfg.wipe_file_path() is None + assert cfg.display_target() == "postgresql+asyncpg://user:pass@localhost:5432/persist_db" + + +def test_persist_config_keeps_sqlite_db_path_compatible() -> None: + cfg = PersistConfig(backend="sqlite", db_path="runtime/test.db") + + assert cfg.resolved_url() == "sqlite+aiosqlite:///runtime/test.db" + assert cfg.resolved_db_path() == "runtime/test.db" + assert cfg.wipe_file_path() == "runtime/test.db" + assert cfg.display_target() == "runtime/test.db" + + +@pytest.mark.asyncio +async def test_sqlalchemy_backend_sqlite_url_creates_schema_and_persists(tmp_path: Path) -> None: + register_test_domain() + + db_path = tmp_path / "sqlalchemy_persist.db" + cfg = PersistConfig(backend="sqlalchemy", url=f"sqlite+aiosqlite:///{db_path}") + backend = create_persistence_backend(cfg) + + assert isinstance(backend, SQLAlchemyPersistenceBackend) + + await backend.initialize() + + local_collection = DataCollection("local", scope="scope-a", persistence=backend, auto_persist=False) + bindings = BindingManager(backend, auto_persist=False, scope="scope-a") + + node = build_project_node("node-1", "project-1", name="Project 1", code="P1") + await local_collection.add(node) + await local_collection.persist() + await bindings.bind("test_project", node.node_id, "remote-1") + await bindings.persist() + + reloaded = DataCollection("local", scope="scope-a", persistence=backend, auto_persist=False) + await reloaded.load_from_persistence() + loaded_nodes = reloaded.filter(node_type="test_project") + assert len(loaded_nodes) == 1 + assert loaded_nodes[0].data_id == "project-1" + + reloaded_bindings = BindingManager(backend, auto_persist=False, scope="scope-a") + await reloaded_bindings.load_from_persistence("test_project") + assert await reloaded_bindings.get_remote_id("test_project", "node-1") == "remote-1" + + columns, rows = PersistenceBackend.query_records( + backend="sqlite", + db_path=str(db_path), + sql="PRAGMA index_list('nodes')", + ) + assert columns == ["seq", "name", "unique", "origin", "partial"] + assert any(row["name"] == "idx_nodes_scope_type_node_id" for row in rows) + + sqlalchemy_columns, sqlalchemy_rows = PersistenceBackend.query_records( + backend="sqlalchemy", + url=f"sqlite+aiosqlite:///{db_path}", + sql="SELECT scope, collection_id, COUNT(1) AS count FROM nodes GROUP BY scope, collection_id ORDER BY scope, collection_id", + ) + assert sqlalchemy_columns == ["scope", "collection_id", "count"] + assert sqlalchemy_rows == [{"scope": "scope-a", "collection_id": "local", "count": 1}] + + sqlalchemy_summary = PersistenceBackend.records_summary( + backend="sqlalchemy", + url=f"sqlite+aiosqlite:///{db_path}", + ) + assert any(item["name"] == "nodes" for item in sqlalchemy_summary["tables"]) + assert any(item["name"] == "bindings" for item in sqlalchemy_summary["tables"]) + + await backend.close() + + +@pytest.mark.asyncio +async def test_sqlalchemy_backend_bulk_replace_chunks_delete_batches(tmp_path: Path) -> None: + register_test_domain() + + db_path = tmp_path / "sqlalchemy_chunked_persist.db" + cfg = PersistConfig(backend="sqlalchemy", url=f"sqlite+aiosqlite:///{db_path}") + backend = create_persistence_backend(cfg) + + assert isinstance(backend, SQLAlchemyPersistenceBackend) + + backend.bulk_delete_batch_size = 2 + await backend.initialize() + + collection = DataCollection("local", scope="scope-a", persistence=backend, auto_persist=False) + for index in range(5): + await collection.add(build_project_node(f"node-{index}", f"project-{index}", name=f"Project {index}", code=f"P{index}")) + await collection.persist() + + refreshed = DataCollection("local", scope="scope-a", persistence=backend, auto_persist=False) + for index in range(5): + await refreshed.add( + build_project_node(f"node-{index}", f"project-{index}-updated", name=f"Project {index} Updated", code=f"Q{index}") + ) + await refreshed.persist() + + reloaded = DataCollection("local", scope="scope-a", persistence=backend, auto_persist=False) + await reloaded.load_from_persistence() + loaded_nodes = sorted(reloaded.filter(node_type="test_project"), key=lambda item: item.node_id) + + assert len(loaded_nodes) == 5 + assert [node.data_id for node in loaded_nodes] == [f"project-{index}-updated" for index in range(5)] + + await backend.close() + + +def test_sqlalchemy_backend_mysql_ddl_uses_varchar_for_keyed_columns() -> None: + backend = SQLAlchemyPersistenceBackend("mysql+aiomysql://root:123456@127.0.0.1/ecm_sync_system?charset=utf8mb4") + + nodes_sql = str(CreateTable(backend.nodes).compile(dialect=mysql.dialect())) + bindings_sql = str(CreateTable(backend.bindings).compile(dialect=mysql.dialect())) + + assert "collection_id VARCHAR(255) NOT NULL" in nodes_sql + assert "scope VARCHAR(255) NOT NULL" in nodes_sql + assert "node_id VARCHAR(255) NOT NULL" in nodes_sql + assert "node_type VARCHAR(128) NOT NULL" in nodes_sql + assert "data_id VARCHAR(255)" in nodes_sql + assert "bind_data_id VARCHAR(255)" in nodes_sql + assert "TEXT NOT NULL" not in nodes_sql + + assert "scope VARCHAR(255) NOT NULL" in bindings_sql + assert "node_type VARCHAR(128) NOT NULL" in bindings_sql + assert "local_id VARCHAR(255) NOT NULL" in bindings_sql + assert "remote_id VARCHAR(255)" in bindings_sql \ No newline at end of file diff --git a/tests/unit/test_summary_report.py b/tests/unit/test_summary_report.py index b2946a3..c4f77e3 100644 --- a/tests/unit/test_summary_report.py +++ b/tests/unit/test_summary_report.py @@ -6,7 +6,7 @@ from io import StringIO import pytest from sync_state_machine.pipeline.summary_report import print_pipeline_summary -from sync_state_machine.logging import get_scope_logger +from sync_state_machine.logging import clear_scope_display_names, get_scope_logger, set_scope_display_name class _FakeBindingManager: @@ -79,7 +79,57 @@ async def test_pipeline_summary_omits_diff_schemas_column() -> None: assert "None/'2025-11-14'" in output +@pytest.mark.asyncio +async def test_pipeline_summary_only_prefixes_headers_for_scoped_logger() -> None: + clear_scope_display_names() + set_scope_display_name("project_id:p1", "project:Alpha") + stream = StringIO() + base_logger = logging.getLogger("sync_state_machine.tests.summary_report_scoped") + base_logger.handlers.clear() + base_logger.setLevel(logging.INFO) + base_logger.propagate = False + + handler = logging.StreamHandler(stream) + handler.setFormatter(logging.Formatter("%(message)s")) + base_logger.addHandler(handler) + + try: + scoped_logger = get_scope_logger("project_id:p1", "tests.summary_report_scoped") + await print_pipeline_summary( + logger=scoped_logger, + node_types=["project"], + binding_manager=_FakeBindingManager(), + local_datasource=_FakeDataSource(), + remote_datasource=_FakeDataSource(), + local_collection=_FakeCollection(), + remote_collection=_FakeCollection(), + consistency_by_type={ + "project": { + "checked_pairs": 1, + "mismatch_count": 0, + "fields": [], + } + }, + stats={ + "loaded": 1, + "synced": 0, + "failed": 0, + "post_check_passed": True, + }, + ) + finally: + base_logger.removeHandler(handler) + clear_scope_display_names() + + lines = [line for line in stream.getvalue().splitlines() if line.strip()] + assert any(line.startswith("[project:Alpha] 📊 Local Summary") for line in lines) + assert any(line == "Node Type Bound/Rec/Load Create(S/F/T) Update(S/F/T) Delete(S/F/T) Consistent(S/F/T) " for line in lines) + assert not any(line.startswith("[project:Alpha] Node Type") for line in lines) + assert not any(line.startswith("[project:Alpha] TOTAL") for line in lines) + + def test_scope_logger_prefixes_messages() -> None: + clear_scope_display_names() stream = StringIO() base_logger = logging.getLogger("sync_state_machine.tests.scope_logger") base_logger.handlers.clear() @@ -96,4 +146,27 @@ def test_scope_logger_prefixes_messages() -> None: finally: base_logger.removeHandler(handler) - assert stream.getvalue().strip() == "[project_id:p1] Strategy start: project" \ No newline at end of file + assert stream.getvalue().strip() == "[project_id:p1] Strategy start: project" + + +def test_scope_logger_uses_registered_display_name() -> None: + clear_scope_display_names() + set_scope_display_name("project_id:p1", "project:Alpha") + stream = StringIO() + base_logger = logging.getLogger("sync_state_machine.tests.scope_logger_display") + base_logger.handlers.clear() + base_logger.setLevel(logging.INFO) + base_logger.propagate = False + + handler = logging.StreamHandler(stream) + handler.setFormatter(logging.Formatter("%(message)s")) + base_logger.addHandler(handler) + + try: + scoped_logger = get_scope_logger("project_id:p1", "tests.scope_logger_display") + scoped_logger.info("Strategy start: project") + finally: + base_logger.removeHandler(handler) + clear_scope_display_names() + + assert stream.getvalue().strip() == "[project:Alpha] Strategy start: project" \ No newline at end of file