From 21a19ab374ccbbb154b55ec47a77720e8a687ae2 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 19 Aug 2026 12:12:15 -0400 Subject: [PATCH] refactor: converge the missing-column warning into the schema module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两个 recorder 里逐字重复的 `_missing_columns_message` 收敛为 `schema.py` 的 `missing_columns_warning(backend, missing, *, alien_table)`。这条消息拼的是 给人执行的 DDL,与库自己执行的 ALTER 必须同源——留在两个 recorder 里等于在 单一事实源上开了个口子,而 Task 7 的文档还要引用这个消息格式。 纯收敛,行为零变化: 两端语句仍分别取自各自的 `SQLITE_BACKFILL` / `PG_BACKFILL` (函数内不硬编码任何 DDL 文本),措辞、标点与换行逐字保留。已用改动前后的两份 实现对 16 组入参(4 种缺列组合 × alien 两态 × 两后端)逐串比对,输出完全相同。 顺带把 postgres.py 从 281 行降到 250、sqlite.py 降到 157。 --- src/polygateway/telemetry/postgres.py | 49 ++++----------------- src/polygateway/telemetry/schema.py | 61 +++++++++++++++++++++++++++ src/polygateway/telemetry/sqlite.py | 50 ++++------------------ 3 files changed, 79 insertions(+), 81 deletions(-) diff --git a/src/polygateway/telemetry/postgres.py b/src/polygateway/telemetry/postgres.py index 4e270df..f26b223 100644 --- a/src/polygateway/telemetry/postgres.py +++ b/src/polygateway/telemetry/postgres.py @@ -21,7 +21,13 @@ from typing import TYPE_CHECKING from loguru import logger -from polygateway.telemetry.schema import COLUMNS, PG_BACKFILL, PG_DDL, insert_sql +from polygateway.telemetry.schema import ( + COLUMNS, + PG_BACKFILL, + PG_DDL, + insert_sql, + missing_columns_warning, +) if TYPE_CHECKING: import asyncpg @@ -35,44 +41,6 @@ _EXISTING_COLUMNS = ( "WHERE attrelid = to_regclass('llm_calls') AND attnum > 0 AND NOT attisdropped" ) -# 缺列 warning 要打印可直接执行的补列语句,与库内 ALTER 同源(不许两份) -_BACKFILL_STATEMENTS = dict(PG_BACKFILL) - - -def _missing_columns_message(missing: list[str], *, alien_table: bool) -> str: - """拼 manual 档的缺列告警: 逐列点名 + 讲清后果 + 给出可直接执行的 SQL。 - - 只说"缺列"是不够的: 静默丢维度的后果是多租户账目全归空串且无任何报错, - 看告警的人必须一眼看到丢的是哪几个维度、以及怎么补。 - - Args: - missing: 缺失的列名(按 `COLUMNS` 保序)。 - alien_table: 连主键列 `call_id` 都没有——该表多半不是本库的 `llm_calls`。 - - Returns: - 单条 warning 的完整文本(库只在准备期发一次,不逐行发)。 - """ - statements = [ - f"{_BACKFILL_STATEMENTS[column]};" for column in missing if column in _BACKFILL_STATEMENTS - ] - unknown = [column for column in missing if column not in _BACKFILL_STATEMENTS] - if unknown: - # 这些列本库从未经 ALTER 补过(建表即有),给不出单条 ALTER,指向完整脚本 - statements.append( - f"-- 另缺 {', '.join(unknown)};完整建表脚本见 " - 'polygateway.telemetry_schema_sql("postgres")' - ) - head = ( - "Postgres 遥测表 llm_calls 缺主键列 call_id,很可能不是本库的遥测表" - "(库不做二次判定,仍照常尝试写入)" - if alien_table - else "Postgres 遥测表 llm_calls 缺列,且 auto_migrate=False(库不发任何 DDL)" - ) - return ( - f"{head};以下维度不会被记录: {', '.join(missing)}。" - "补列请自行执行(建议挑低峰,ALTER 取 ACCESS EXCLUSIVE 锁):\n" + "\n".join(statements) - ) - class PostgresRecorder: """TelemetryRecorder 端口的 Postgres 实现;asyncpg 原生异步,无线程桥接。""" @@ -231,7 +199,8 @@ class PostgresRecorder: if missing: # 单参数传入: 补列 SQL 里带 `'{}'::jsonb` 字面量,拼进 format 模板会被当占位符 logger.warning( - "{}", _missing_columns_message(missing, alien_table="call_id" not in existing) + "{}", + missing_columns_warning("postgres", missing, alien_table="call_id" not in existing), ) return effective diff --git a/src/polygateway/telemetry/schema.py b/src/polygateway/telemetry/schema.py index 274b5aa..a45c98e 100644 --- a/src/polygateway/telemetry/schema.py +++ b/src/polygateway/telemetry/schema.py @@ -186,6 +186,67 @@ def insert_sql(backend: str, columns: Sequence[str]) -> str: return f"INSERT INTO {TABLE} ({names}) VALUES ({placeholders}) ON CONFLICT DO NOTHING" +# 缺列告警要打印的补列语句: 库内执行的那份怎么写,打印给人的就怎么写(同源不许漂移)。 +# SQLite 侧常量只有列定义,故在此按 TABLE 拼成整条 ALTER;PG 侧常量本就是整条语句。 +_ALTER_BY_BACKEND = { + "sqlite": { + column: f"ALTER TABLE {TABLE} ADD COLUMN {column} {decl}" + for column, decl in SQLITE_BACKFILL + }, + "postgres": dict(PG_BACKFILL), +} + +_BACKEND_LABELS = {"sqlite": "SQLite", "postgres": "Postgres"} + +# PG 的 ALTER 取 ACCESS EXCLUSIVE 锁,执行时机得由 DBA 自己挑;SQLite 是下游本地文件,无此顾虑 +_EXECUTION_NOTES = {"sqlite": "", "postgres": "(建议挑低峰,ALTER 取 ACCESS EXCLUSIVE 锁)"} + + +def missing_columns_warning(backend: str, missing: Sequence[str], *, alien_table: bool) -> str: + """拼 manual 档的缺列告警: 逐列点名 + 讲清后果 + 给出可直接执行的 SQL。 + + 只说"缺列"是不够的: 静默丢维度的后果是多租户账目全归空串且无任何报错, + 看告警的人必须一眼看到丢的是哪几个维度、以及怎么补。 + + **住在本模块而不是两个 recorder 里**: 这条消息拼的是给人执行的 DDL,与库自己 + 执行的 ALTER 必须同源——本模块存在的全部理由就是不许这两者漂移。 + + Args: + backend: `"sqlite"` 或 `"postgres"`。 + missing: 缺失的列名(按 `COLUMNS` 保序)。 + alien_table: 连主键列 `call_id` 都没有——该表多半不是本库的 `llm_calls`。 + + Returns: + 单条 warning 的完整文本(库只在准备期发一次,不逐行发)。 + + Raises: + ValueError: backend 不在取值域内。 + """ + if backend not in _BACKENDS: + raise ValueError(f"未知遥测后端 {backend!r}: 只支持 {list(_BACKENDS)}") + alters = _ALTER_BY_BACKEND[backend] + selected = tuple(missing) + statements = [f"{alters[column]};" for column in selected if column in alters] + unknown = [column for column in selected if column not in alters] + if unknown: + # 这些列本库从未经 ALTER 补过(建表即有),给不出单条 ALTER,指向完整脚本 + statements.append( + f"-- 另缺 {', '.join(unknown)};完整建表脚本见 " + f'polygateway.telemetry_schema_sql("{backend}")' + ) + label = _BACKEND_LABELS[backend] + head = ( + f"{label} 遥测表 {TABLE} 缺主键列 call_id,很可能不是本库的遥测表" + "(库不做二次判定,仍照常尝试写入)" + if alien_table + else f"{label} 遥测表 {TABLE} 缺列,且 auto_migrate=False(库不发任何 DDL)" + ) + return ( + f"{head};以下维度不会被记录: {', '.join(selected)}。" + f"补列请自行执行{_EXECUTION_NOTES[backend]}:\n" + "\n".join(statements) + ) + + def telemetry_schema_sql(backend: str) -> str: """返回可直接粘进迁移文件的完整脚本(建表 + 各补列语句 + 注释)。 diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 604a676..15f0b18 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -22,46 +22,13 @@ from pathlib import Path from loguru import logger -from polygateway.telemetry.schema import COLUMNS, SQLITE_BACKFILL, SQLITE_DDL, insert_sql - -# 缺列 warning 要打印可直接执行的补列语句,列定义与库内 ALTER 同源(不许两份) -_BACKFILL_DECLS = dict(SQLITE_BACKFILL) - - -def _missing_columns_message(missing: list[str], *, alien_table: bool) -> str: - """拼 manual 档的缺列告警: 逐列点名 + 讲清后果 + 给出可直接执行的 SQL。 - - 只说"缺列"是不够的: 静默丢维度的后果是多租户账目全归空串且无任何报错, - 看告警的人必须一眼看到丢的是哪几个维度、以及怎么补。 - - Args: - missing: 缺失的列名(按 `COLUMNS` 保序)。 - alien_table: 连主键列 `call_id` 都没有——该表多半不是本库的 `llm_calls`。 - - Returns: - 单条 warning 的完整文本(库只在准备期发一次,不逐行发)。 - """ - statements = [ - f"ALTER TABLE llm_calls ADD COLUMN {column} {_BACKFILL_DECLS[column]};" - for column in missing - if column in _BACKFILL_DECLS - ] - unknown = [column for column in missing if column not in _BACKFILL_DECLS] - if unknown: - # 这些列本库从未经 ALTER 补过(建表即有),给不出单条 ALTER,指向完整脚本 - statements.append( - f"-- 另缺 {', '.join(unknown)};完整建表脚本见 " - 'polygateway.telemetry_schema_sql("sqlite")' - ) - head = ( - "SQLite 遥测表 llm_calls 缺主键列 call_id,很可能不是本库的遥测表" - "(库不做二次判定,仍照常尝试写入)" - if alien_table - else "SQLite 遥测表 llm_calls 缺列,且 auto_migrate=False(库不发任何 DDL)" - ) - return f"{head};以下维度不会被记录: {', '.join(missing)}。补列请自行执行:\n" + "\n".join( - statements - ) +from polygateway.telemetry.schema import ( + COLUMNS, + SQLITE_BACKFILL, + SQLITE_DDL, + insert_sql, + missing_columns_warning, +) class SQLiteRecorder: @@ -138,7 +105,8 @@ class SQLiteRecorder: if missing: # 单参数传入: 补列 SQL 里带 `'{}'` 字面量,拼进 format 模板会被当占位符 logger.warning( - "{}", _missing_columns_message(missing, alien_table="call_id" not in existing) + "{}", + missing_columns_warning("sqlite", missing, alien_table="call_id" not in existing), ) def _backfill_columns(self, existing: set[str]) -> None: