From e17e1067a1b77ba3868e771b6e133f6cdd719b3e Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 19 Aug 2026 12:23:07 -0400 Subject: [PATCH] feat: derive the schema mode from the telemetry backend --- .env.example | 9 ++++++ src/polygateway/config.py | 51 +++++++++++++++++++++++++----- tests/unit/test_config.py | 65 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 118 insertions(+), 7 deletions(-) diff --git a/.env.example b/.env.example index 14b7c58..08ca2b9 100644 --- a/.env.example +++ b/.env.example @@ -56,6 +56,15 @@ PGW_BREAKER_BACKEND=memory # memory | redis PGW_CACHE_BACKEND=none # redis | memory | none(必填,显式优于隐式) PGW_TELEMETRY_BACKEND=none # sqlite | postgres | none(必填) # PGW_TELEMETRY_SQLITE_PATH=logs/telemetry.db # sqlite 时必填 +# PGW_TELEMETRY_SCHEMA_MODE=manual # auto | manual;三态: 不设 = 按后端派生(sqlite→auto、postgres→manual), +# # 显式设置则两侧都可覆盖。auto = 库给已存在的旧表自动 ALTER 补列; +# # manual = 库不发 ALTER,只 warning 点名缺列并打印可执行 SQL, +# # 按现有列裁剪 INSERT 继续写(遥测不会因缺列而全线丢失)。 +# # 缺省为何不对称: postgres 是共享生产表,ALTER 取 ACCESS EXCLUSIVE 锁, +# # 会排在长事务后阻塞该表其后的所有查询,而遥测是业务路径上的内联 await; +# # 且这类部署有 DBA、有迁移工具、讲最小权限,DDL 该由他们择时执行。 +# # sqlite 则是下游自己的本地文件(runs/*.db):没有 DBA、没有迁移工具、 +# # 没有第二个系统碰它,ALTER 是毫秒级元数据操作,强加手工 SQL 步骤是净损失。 # PGW_TELEMETRY_PG_DSN=postgresql://user:pass@host:5432/polygateway # postgres 时必填;严禁指向在用业务库(实验室约定: 专用库 polygateway) # PGW_PRICING_PATH=config/prices.json # 可选: {"": {"input_per_1m": x, "output_per_1m": y}};缺省 cost 恒 None # # 可选第三档 "cached_input_per_1m": z —— 供应商 prompt cache 命中部分的单价; diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 66c776d..7ab8e8b 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -55,6 +55,9 @@ _LIMITER_BACKENDS = frozenset({"memory", "redis"}) _BREAKER_BACKENDS = frozenset({"memory", "redis"}) _CACHE_BACKENDS = frozenset({"redis", "memory", "none"}) _TELEMETRY_BACKENDS = frozenset({"sqlite", "postgres", "none"}) +# 遥测 schema 档位(issue #13): auto 允许 recorder 给旧表 ALTER 补列,manual 不发 DDL +_SCHEMA_MODES = frozenset({"auto", "manual"}) +_SCHEMA_MODE_KEY = "PGW_TELEMETRY_SCHEMA_MODE" _REDIS_DEPENDENT_BACKENDS = ("limiter_backend", "breaker_backend", "cache_backend") # 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源) _DEFAULT_STALL_WINDOW_S = 300.0 @@ -131,8 +134,10 @@ class GatewaySettings: telemetry_backend: str telemetry_sqlite_path: str | None telemetry_pg_dsn: str | None - # 是否允许 recorder 给已存在的旧表自动 ALTER 补列(issue #13); - # 派生规则只写在 `_load_pgw` 一处,不与 recorder 的类签名漂移 + # 是否允许 recorder 给已存在的旧表自动 ALTER 补列(issue #13);env 的三态 + # 派生只写在 `_load_schema_mode` 一处,不与 recorder 的类签名漂移。 + # backend=none 时恒 False 这条跨字段不变量则由 `_validate_telemetry` + # 把关,对直接构造与 `dataclasses.replace` 同样生效 telemetry_auto_migrate: bool redis_url: str | None pricing_path: str | None @@ -214,7 +219,14 @@ class GatewaySettings: 剥而不是拒: 两条装配路对同一 DSN 应产出同一结果。但不静默——`from_env` 那条路在 `_load_pg_dsn` 就剥干净了,能走到这里的只有手工构造的调用方, 他有权知道库动了他给的值。 + + `telemetry_auto_migrate` 同理归一化而非报错: backend=none 时根本没有 + recorder 消费它,True 是个自相矛盾却无害的状态。`from_env` 那条路的派生 + 已经给出 False,归一化是为了直接构造与 `dataclasses.replace` 也一致—— + 不变量挂在构造期,才不用每加一个装配工厂就多一处要同步。 """ + if self.telemetry_backend == "none" and self.telemetry_auto_migrate: + object.__setattr__(self, "telemetry_auto_migrate", False) if self.telemetry_backend == "sqlite" and not self.telemetry_sqlite_path: raise ValueError("telemetry_backend=sqlite 时必须提供 telemetry_sqlite_path") if self.telemetry_backend != "postgres": @@ -437,6 +449,7 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: redis_url = env.get("REDIS_URL") or None if "redis" in (limiter_backend, breaker_backend) and redis_url is None: raise ValueError("缺关键配置: 限流/熔断后端取 redis 需设置 REDIS_URL") + auto_migrate = _load_schema_mode(env, telemetry_backend) return { "limiter_backend": limiter_backend, "breaker_backend": breaker_backend, @@ -447,11 +460,7 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: if telemetry_backend == "sqlite" else None, "telemetry_pg_dsn": _load_pg_dsn(env) if telemetry_backend == "postgres" else None, - # 按后端不对称派生(issue #13): SQLite 是下游自己的本地文件(无 DBA、无迁移 - # 工具),补列是毫秒级元数据操作;PG 是共享生产表,ALTER 取 ACCESS EXCLUSIVE - # 锁会阻塞该表其后所有查询,而遥测是业务路径上的内联 await。 - # backend=none 时无 recorder 消费该值,派生结果恒 False。 - "telemetry_auto_migrate": telemetry_backend == "sqlite", + "telemetry_auto_migrate": auto_migrate, "redis_url": redis_url, "pricing_path": env.get("PGW_PRICING_PATH") or None, "structured_max_retries": _load_structured_retries(env), @@ -459,6 +468,34 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: } +def _load_schema_mode(env: Mapping[str, str], telemetry_backend: str) -> bool: + """把 `PGW_TELEMETRY_SCHEMA_MODE` 的三态解成 `telemetry_auto_migrate`(issue #13)。 + + 三态: 键未设 → 按后端**不对称**派生;显式 auto/manual → 两侧都可覆盖。 + 不对称的理由是两个后端的风险量级不同: SQLite 是下游自己的本地文件(没有 + DBA、没有迁移工具、没有第二个系统碰它),ALTER 是毫秒级元数据操作,要求 + 手工跑 SQL 是给零运维场景强加运维步骤;PG 是共享的生产表,ALTER 取 + ACCESS EXCLUSIVE 锁会排在长事务后阻塞该表其后的所有查询,而遥测是业务 + 路径上的内联 await。 + + `_load_choice` 带 default,不能直接用来读这个键——default 会把"未设"和 + "设成默认值"抹平成同一种,三态就塌回两态,后端派生也就再没机会生效。故 + 先用 `_first` 探"设没设",确认设了才交给 `_load_choice` 做值域校验(错误 + 信息点出 env 键名这件事仍由它负责)。 + + Args: + env: 已合并的环境映射。 + telemetry_backend: 已校验过值域的遥测后端名。 + + Returns: + recorder 是否获准给旧表自动 ALTER 补列;backend=none 时无人消费, + 构造期守卫会再把它归一化为 False。 + """ + if _first(env, _SCHEMA_MODE_KEY) is None: + return telemetry_backend == "sqlite" + return _load_choice(env, _SCHEMA_MODE_KEY, _SCHEMA_MODES, "auto") == "auto" + + def _strip_dsn_driver(dsn: str) -> str: """剥 SQLAlchemy 风格的 `+driver` 后缀(asyncpg 不认);已干净的原样返回。""" scheme, sep, rest = dsn.partition("://") diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 0532dfc..0bf1d27 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -331,6 +331,61 @@ class TestAssemblyGuards: assert GatewaySettings.from_env("LLM", env=env_ok).backpressure.stall_window_s == 60.0 +class TestTelemetrySchemaMode: + """PGW_TELEMETRY_SCHEMA_MODE 三态(issue #13 设计 §4.1)。 + + 键未设时按后端**不对称**派生: SQLite 是下游自己的本地文件(没有 DBA、 + 没有迁移工具、没有第二个系统碰它),补列是毫秒级元数据操作,故默认 auto; + PG 是共享生产表,ALTER 取 ACCESS EXCLUSIVE 锁会阻塞该表其后的所有查询, + 而遥测是业务路径上的内联 await,故默认 manual。显式设置两侧都可覆盖—— + "可覆盖"正是三态相对两态多出来的那一态,派生本身盖不住它。 + """ + + def _sqlite_env(self, **overrides): + return _env( + PGW_TELEMETRY_BACKEND="sqlite", + PGW_TELEMETRY_SQLITE_PATH="logs/telemetry.db", + **overrides, + ) + + def _pg_env(self, **overrides): + return _env( + PGW_TELEMETRY_BACKEND="postgres", + PGW_TELEMETRY_PG_DSN="postgresql://u:p@h:5432/polygateway", + **overrides, + ) + + def test_unset_key_derives_auto_for_sqlite(self): + s = GatewaySettings.from_env("LLM", env=self._sqlite_env()) + assert s.telemetry_auto_migrate is True + + def test_unset_key_derives_manual_for_postgres(self): + s = GatewaySettings.from_env("LLM", env=self._pg_env()) + assert s.telemetry_auto_migrate is False + + def test_unset_key_derives_manual_for_none_backend(self): + """backend=none 无 recorder 消费该字段,派生结果必须是 False 而非 sqlite 那档。""" + s = GatewaySettings.from_env("LLM", env=_env()) + assert s.telemetry_auto_migrate is False + + def test_explicit_manual_overrides_sqlite_default(self): + s = GatewaySettings.from_env( + "LLM", env=self._sqlite_env(PGW_TELEMETRY_SCHEMA_MODE="manual") + ) + assert s.telemetry_auto_migrate is False + + def test_explicit_auto_overrides_postgres_default(self): + s = GatewaySettings.from_env("LLM", env=self._pg_env(PGW_TELEMETRY_SCHEMA_MODE="auto")) + assert s.telemetry_auto_migrate is True + + def test_invalid_mode_rejected_naming_the_env_key(self): + """报错须点出 env 键名: 这条路的调用方看得懂的是键名,不是字段名。""" + with pytest.raises(ValueError, match="PGW_TELEMETRY_SCHEMA_MODE"): + GatewaySettings.from_env( + "LLM", env=self._sqlite_env(PGW_TELEMETRY_SCHEMA_MODE="enabled") + ) + + class TestOcrSettings: """M3 OcrSettings(设计 §3.4): 复用 GatewaySettings,无 OCR 专用键。""" @@ -555,6 +610,16 @@ class TestCrossFieldInvariants: with pytest.raises(ValueError, match="telemetry_pg_dsn"): dataclasses.replace(base, telemetry_backend="postgres") + def test_none_backend_forces_auto_migrate_off(self): + """backend=none 时没有 recorder 消费该字段,True 是自相矛盾的状态(issue #13)。 + + env 路的派生已给出 False,但直接构造与 dataclasses.replace 这两条同等 + 官方的装配路仍能把 True 传进来——不变量归位到构造期,三条路才一致。 + """ + base = self._base() # telemetry_backend="none" + replaced = dataclasses.replace(base, telemetry_auto_migrate=True) + assert replaced.telemetry_auto_migrate is False + # —— 标量域 —— def test_negative_structured_retries_rejected(self):