331 lines
14 KiB
Python
331 lines
14 KiB
Python
#!/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())
|