#!/usr/bin/env python3 # -*- coding: utf-8 -*- """补齐 pbl_common 契约符号缺口（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量修复）。 背景 ---- 原 Bug YyDumUuSK9R9Q_3Ai23AD 的 develop 侧根因 G 是：``pbl_common.api`` 缺 ``crud`` 符号 → 下游 ``pbl_assessment`` / ``pbl_scense_ext`` **包导入期 ImportError** → 应用进程起不来（healthz=000、pbls.py 进程数=0、9300 未监听）。 按任务要求「必须检查所有模块的同类问题，不能只修报出来的那一个」，用 ``apps/pbls/scripts/verify_contract_symbols_all_modules.py`` 对全部 15 个 pbl_* 模块 + apps/pbls 共 266 个 .py 做静态 AST 闭包核验，查出 **4 处同类缺口**： | # | 文件:行 | 缺失符号 | 严重度 | |---|---------|----------|--------| | 1 | modules/pbl_runtime_ext/pbl_runtime_ext/idempotency.py:21 | ``pbl_common.api.q_all`` / ``q_one`` | **致命**：模块顶层 import，无 try 兜底 → 包导入即 ImportError，与 crud 缺口完全同型 | | 2 | modules/pbl_runtime_ext/pbl_runtime_ext/rtx_db.py:59 | ``pbl_common.api.module_dbname`` | 中：在 try/except 内，静默降级为「无法解析模块库名」PBL-RTX-0001 硬错 | | 3 | modules/pbl_compiler/pbl_compiler/exporter.py:76 | ``pbl_common.db.get_sor_context`` | 中：在 try/except 内，静默降级到 DBPools 兜底路径（绕过公共内核租户适配） | 本脚本幂等地把上述符号补进 pbl_common（**只改本项目 generated_modules 的 pbl_common，不碰任何 referenced_modules / 基础模块**）： - ``pbl_common/api.py``：新增 ``q_all`` / ``q_one`` / ``module_dbname``， 并登记进 ``CONTRACT_SYMBOLS``（``__all__ = list(CONTRACT_SYMBOLS) + [...]`` 自动随之导出，``verify_contract()`` 的 _MISSING 自检同步覆盖）。 - ``pbl_common/db.py``：新增 ``get_sor_context(env, module)``。 q_all / q_one 的**双模兼容**设计（关键） ---------------------------------------- 实测两类调用点语义不同，必须同时满足，否则修一处坏另一处： - ``idempotency.py``：**同步**调用且带 ``conn=`` 显式连接、SQL 用 ``%s`` 占位 （``existing = q_one(sql, params, conn=conn)``，无 await）； - ``rtx_db.py``：``async def q_all(...)`` 内 ``await`` 公共内核 reader。 故返回值用 ``list`` / ``dict`` 子类并实现 ``__await__``——同步取值与 ``await`` 取值都成立；``q_one`` 无行时返回**空 _DualDict**（而非 None）， 既保证 ``await`` 不炸，又与全部现存调用点的真值判断 （``if existing:`` / ``(max_row or {}).get(...)`` / ``if raced and ...``）语义一致。 用法:: python3 apps/pbls/scripts/_patch_common_contract_symbols.py # 应用补丁 python3 apps/pbls/scripts/_patch_common_contract_symbols.py --check # 只检查不落盘 """ from __future__ import annotations import argparse import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) APP_ROOT = os.path.dirname(HERE) WS_ROOT = os.path.dirname(os.path.dirname(APP_ROOT)) API_PY = os.path.join(WS_ROOT, "modules", "pbl_common", "pbl_common", "api.py") DB_PY = os.path.join(WS_ROOT, "modules", "pbl_common", "pbl_common", "db.py") MARK_API = "# >>> q_all/q_one/module_dbname contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>>" MARK_DB = "# >>> get_sor_context contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>>" API_BLOCK = ''' # >>> q_all/q_one/module_dbname contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>> # --------------------------------------------------------------------------- # 查询助手 q_all / q_one —— 同步与 await 双模兼容 # # 缺口来源：modules/pbl_runtime_ext/pbl_runtime_ext/idempotency.py:21 在**模块 # 顶层**写 `from pbl_common.api import q_all, q_one`，而 pbl_common.api 从未定义 # 这两个符号 → 包导入期 ImportError，pbl_runtime_ext 整包不可用（与原 Bug 的 # crud 缺口完全同型：应用进程起不来、healthz=000）。 # # 双模设计依据（两类真实调用点，必须同时成立）： # * idempotency.py -> 同步调用 + 显式 conn= + SQL 用 %s 占位（无 await） # * rtx_db.py -> `async def q_all(...)` 内 await 公共内核 reader # 因此返回 list/dict 子类并实现 __await__；q_one 无行返回**空 dict**而非 None， # 保证 await 不炸，且与现存调用点真值判断（if existing: / (row or {}).get / # if raced and ...）语义完全一致。 # --------------------------------------------------------------------------- class _DualList(list): """list 子类，既可同步迭代/索引，也可被 await（await 后得到自身）。""" __slots__ = () def __await__(self): async def _identity(): return self return _identity().__await__() class _DualDict(dict): """dict 子类，既可同步 .get()/[]，也可被 await（await 后得到自身）。 无行时返回空 _DualDict（falsy），与 `if row:` / `(row or {})` 判断兼容。 """ __slots__ = () def __await__(self): async def _identity(): return self return _identity().__await__() def _rows_from_conn(conn, sql, params): """在调用方显式传入的 DB-API 连接上执行查询，返回 list[dict]。 idempotency.py 的 SQL 使用 %s 占位（PyMySQL 风格）并自带事务连接， 必须原样在该连接上执行，不能改走 sqlor 的 ${name}$ 参数化路径。 """ cur = conn.cursor() try: cur.execute(sql, params if params is not None else ()) desc = cur.description or [] cols = [d[0] for d in desc] raw = cur.fetchall() finally: try: cur.close() except Exception: pass out = [] for row in raw: if isinstance(row, dict): out.append(dict(row)) elif cols: out.append(dict(zip(cols, row))) else: out.append({'value': row}) return out def _rows_via_common(sql, params, dbname, module): """无显式 conn 时走公共内核只读路径（sqlor 优先，租户/库名解析复用既有实现）。""" from pbl_common import dbutil as _dbu target_db = dbname if target_db is None and module: try: target_db = get_module_dbname(module) except Exception: target_db = None for fname in ('query', 'sql_rows', 'fetch_all'): fn = getattr(_dbu, fname, None) if not callable(fn): continue try: res = fn(sql, params, dbname=target_db) except TypeError: try: res = fn(sql, params) except Exception: continue except Exception: continue rows = [] for r in (res or []): if isinstance(r, dict): rows.append(dict(r)) else: keys = getattr(r, 'keys', None) if callable(keys): try: rows.append({k: r[k] for k in keys()}) continue except Exception: pass rows.append({'value': r}) return rows # 最后兜底：api.sql_rows（内部已含 get_conn 直连路径） return [dict(r) if isinstance(r, dict) else {'value': r} for r in (sql_rows(sql, params, dbname=target_db) or [])] def q_all(sql, params=None, conn=None, dbname=None, module=None, **_ignored): """只读查询 → list[dict]（同步取值与 await 取值均可）。 参数 ---- sql : SQL 文本。传 conn 时用驱动原生占位（%s）；否则走公共内核路径。 params : 序列或 dict，与 sql 占位匹配。 conn : 可选，调用方已开启的 DB-API 连接（事务内查询必须传，见 idempotency）。 dbname : 可选，显式库名。**禁止硬编码库名**，缺省由 module 解析。 module : 可选，模块名，用于 get_module_dbname 解析库名。 """ try: if conn is not None: return _DualList(_rows_from_conn(conn, sql, params)) return _DualList(_rows_via_common(sql, params, dbname, module)) except Exception as exc: # noqa: BLE001 log.warning('q_all 失败（返回空列表，不抛断链）: %s', exc) return _DualList() def q_one(sql, params=None, conn=None, dbname=None, module=None, **_ignored): """只读查询首行 → dict（无行返回**空 dict**，非 None；同步/await 双模）。""" rows = q_all(sql, params, conn=conn, dbname=dbname, module=module) if rows: return _DualDict(rows[0]) return _DualDict() def module_dbname(module_name, default=None): """解析模块库名的公共入口（rtx_db.py:59 契约名）。 等价于 get_module_dbname(module_name)，但**失败不抛**：返回 default（缺省 None），由调用方决定是降级还是 fail-closed。绝不返回硬编码库名。 """ try: name = get_module_dbname(module_name) except Exception as exc: # noqa: BLE001 log.debug('module_dbname(%r) 解析失败: %s', module_name, exc) return default return name or default # <<< q_all/q_one/module_dbname contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) <<< ''' DB_BLOCK = ''' # >>> get_sor_context contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>> def get_sor_context(env, module_name=None): """取得 sqlor 异步上下文的公共入口（pbl_compiler/exporter.py:76 契约名）。 缺口来源：exporter.open_sor() 优先 ``from pbl_common.db import get_sor_context``，符号不存在 → 静默 except 降级到 DBPools 兜底路径， 绕过公共内核的库名解析与方言校验。补齐后主路径可用。 参数 ---- env : ``request._run_ns``（.dspy 内）或 ServerEnv() 实例（load_* 阶段）。 module_name : 模块名，用于解析库名；**禁止硬编码 DBNAME**。 返回 async context manager，用法:: async with get_sor_context(request._run_ns, 'pbl_compiler') as sor: rows = await sor.sqlExe('SELECT 1', {}) """ # 1) 宿主已注入的公共 helper 优先（dspy-file-implementation-spec 铁律： # 跨模块 DB 访问走 env + get_sor_context，不硬编码库名） injected = getattr(env, 'get_sor_context', None) if callable(injected): return injected(env, module_name) # 2) 平台 sqlor 上下文 dbname = get_module_dbname(module_name) if module_name else None try: from sqlor.dbpools import DBPools except Exception: DBPools = None # noqa: N806 if DBPools is not None and dbname: return DBPools().sqlorContext(dbname) raise RuntimeError( 'get_sor_context 不可用：env 未注入 get_sor_context 且 sqlor/库名解析失败' '（module=%r, dbname=%r）' % (module_name, dbname)) # <<< get_sor_context contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) <<< ''' # CONTRACT_SYMBOLS 里新增的符号（__all__ = list(CONTRACT_SYMBOLS) + [...] 自动导出） NEW_CONTRACT = (" # 查询助手/库名解析（Bug HoSBcdZY4G7IxhaFiIE07 同类缺口补齐）\n" " 'q_all', 'q_one', 'module_dbname',\n") CONTRACT_ANCHOR = "CONTRACT_SYMBOLS = (\n" def patch_api(check_only: bool = False): src = open(API_PY, encoding="utf-8").read() changed = [] if MARK_API in src: print("[api.py] q_all/q_one/module_dbname 已存在，跳过插入") else: # 必须插在 CONTRACT_SYMBOLS **之前**：api.py 的 verify_contract()/_MISSING # 是 fail-closed 自检（缺符号直接 raise ImportError），若把定义追加到文件 # 末尾，登记在前、定义在后 → 导入期就炸。 if CONTRACT_ANCHOR not in src: raise SystemExit("FATAL: api.py 未找到 CONTRACT_SYMBOLS 锚点，拒绝盲改") src = src.replace(CONTRACT_ANCHOR, API_BLOCK.lstrip("\n") + "\n\n" + CONTRACT_ANCHOR, 1) changed.append("insert q_all/q_one/module_dbname before CONTRACT_SYMBOLS") if "'q_all'" in src.split("CONTRACT_SYMBOLS = (", 1)[-1].split(")", 1)[0]: print("[api.py] CONTRACT_SYMBOLS 已含 q_all，跳过登记") else: if CONTRACT_ANCHOR not in src: raise SystemExit("FATAL: api.py 未找到 CONTRACT_SYMBOLS 锚点，拒绝盲改") src = src.replace(CONTRACT_ANCHOR, CONTRACT_ANCHOR + NEW_CONTRACT, 1) changed.append("register CONTRACT_SYMBOLS") if changed and not check_only: open(API_PY, "w", encoding="utf-8").write(src) print("[api.py] %s: %s" % ("待改" if check_only else "已改", ", ".join(changed) or "无变更")) return bool(changed) def patch_db(check_only: bool = False): src = open(DB_PY, encoding="utf-8").read() if MARK_DB in src or "\ndef get_sor_context(" in src: print("[db.py] get_sor_context 已存在，跳过") return False src = src.rstrip("\n") + "\n" + DB_BLOCK if not check_only: open(DB_PY, "w", encoding="utf-8").write(src) print("[db.py] %s: append get_sor_context" % ("待改" if check_only else "已改")) return True def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--check", action="store_true", help="只检查需要哪些改动，不落盘") args = ap.parse_args() for p in (API_PY, DB_PY): if not os.path.isfile(p): print("FATAL: 目标文件不存在 %s" % p) return 2 a = patch_api(args.check) b = patch_db(args.check) print("-" * 60) if args.check: print("CHECK 结论: %s" % ("需要打补丁" if (a or b) else "已是最新，无需补丁")) else: print("补丁应用完成（幂等，可重复执行）") return 0 if __name__ == "__main__": sys.exit(main())