deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
9050c6c832
commit
35f022146c
330
scripts/_patch_common_contract_symbols.py
Normal file
330
scripts/_patch_common_contract_symbols.py
Normal file
@ -0,0 +1,330 @@
|
||||
#!/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())
|
||||
200
scripts/_selftest_contract_symbols.py
Normal file
200
scripts/_selftest_contract_symbols.py
Normal file
@ -0,0 +1,200 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""契约符号补齐后的行为自测（Bug HoSBcdZY4G7IxhaFiIE07）。
|
||||
|
||||
不依赖数据库/基座：用假连接与假 env 验证 q_all/q_one 的**双模语义**
|
||||
（同步取值 + await 取值）、q_one 无行返回空 dict、conn= 显式连接路径、
|
||||
module_dbname 失败不抛、db.get_sor_context 三级降级顺序。
|
||||
|
||||
用法::
|
||||
PYTHONPATH=modules python3 apps/pbls/scripts/_selftest_contract_symbols.py
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
WS_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(HERE)))
|
||||
MODULES_DIR = os.path.join(WS_ROOT, "modules")
|
||||
# 模块仓库布局是 modules/{模块名}/{模块名}/（包目录=模块名，非 src 布局），
|
||||
# 故需把每个模块仓库根逐个加入 sys.path，`import pbl_common` 才能命中包。
|
||||
for _name in sorted(os.listdir(MODULES_DIR)):
|
||||
_repo = os.path.join(MODULES_DIR, _name)
|
||||
if os.path.isdir(os.path.join(_repo, _name)):
|
||||
sys.path.insert(0, _repo)
|
||||
|
||||
PASS, FAIL = [], []
|
||||
|
||||
|
||||
def check(name, cond, detail=""):
|
||||
(PASS if cond else FAIL).append(name)
|
||||
print(" [%s] %s%s" % ("PASS" if cond else "FAIL", name,
|
||||
(" → " + detail) if detail and not cond else ""))
|
||||
|
||||
|
||||
class FakeCursor:
|
||||
"""DB-API 风格假游标：%s 占位、description、fetchall。"""
|
||||
|
||||
def __init__(self, rows, cols):
|
||||
self._rows, self._cols = rows, cols
|
||||
self.executed = None
|
||||
self.description = [(c,) for c in cols] if cols else []
|
||||
self.closed = False
|
||||
|
||||
def execute(self, sql, params=()):
|
||||
self.executed = (sql, params)
|
||||
assert "%s" in sql or not params, "占位符应原样交给驱动"
|
||||
return None
|
||||
|
||||
def fetchall(self):
|
||||
return self._rows
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
|
||||
|
||||
class FakeConn:
|
||||
def __init__(self, rows, cols):
|
||||
self.cur = FakeCursor(rows, cols)
|
||||
|
||||
def cursor(self):
|
||||
return self.cur
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 68)
|
||||
print("契约符号行为自测（q_all / q_one / module_dbname / get_sor_context）")
|
||||
print("=" * 68)
|
||||
|
||||
from pbl_common import api as A
|
||||
from pbl_common import db as D
|
||||
|
||||
# --- 1. 符号在位（原 Bug 根因 G 的 crud + 本次补齐的 4 个） ---
|
||||
print("\n[1] 符号在位")
|
||||
for sym in ("crud", "q_all", "q_one", "module_dbname"):
|
||||
check("pbl_common.api.%s 可解析" % sym, hasattr(A, sym))
|
||||
check("pbl_common.db.get_sor_context 可解析", hasattr(D, "get_sor_context"))
|
||||
check("CONTRACT_SYMBOLS 含 q_all/q_one/module_dbname",
|
||||
all(s in A.CONTRACT_SYMBOLS for s in ("q_all", "q_one", "module_dbname")))
|
||||
check("__all__ 含 q_all/q_one/module_dbname",
|
||||
all(s in A.__all__ for s in ("q_all", "q_one", "module_dbname")))
|
||||
check("verify_contract() 无缺失符号（_MISSING 为空）",
|
||||
not getattr(A, "_MISSING", None), str(getattr(A, "_MISSING", None)))
|
||||
|
||||
# --- 2. conn= 显式连接路径（idempotency.py 的真实调用形态） ---
|
||||
print("\n[2] q_one(sql, params, conn=conn) 同步调用（idempotency.py:141 形态）")
|
||||
conn = FakeConn([("idem1", "completed", 7)], ["idempotency_id", "state", "client_seq"])
|
||||
row = A.q_one("SELECT idempotency_id, state, client_seq FROM t "
|
||||
"WHERE tenant_id=%s AND client_seq=%s LIMIT 1",
|
||||
("T1", 7), conn=conn)
|
||||
check("同步返回 dict（非 None、非 coroutine）", isinstance(row, dict))
|
||||
check("列名映射正确", row.get("state") == "completed", str(row))
|
||||
check("真值判断 if row: 成立", bool(row) is True)
|
||||
check("参数原样传给驱动", conn.cur.executed[1] == ("T1", 7))
|
||||
check("游标已关闭（无泄漏）", conn.cur.closed is True)
|
||||
|
||||
print("\n[3] q_all(sql, params, conn=conn) 多行")
|
||||
conn2 = FakeConn([(1, "a"), (2, "b")], ["seq", "scope"])
|
||||
rows = A.q_all("SELECT seq, scope FROM t WHERE tenant_id=%s", ("T1",), conn=conn2)
|
||||
check("返回 list 长度 2", isinstance(rows, list) and len(rows) == 2)
|
||||
check("首行 dict 化", rows[0] == {"seq": 1, "scope": "a"}, str(rows[0]))
|
||||
|
||||
# --- 3. await 双模（rtx_db.py 的真实调用形态） ---
|
||||
print("\n[4] await 双模兼容（rtx_db.py:150 形态）")
|
||||
|
||||
async def await_paths():
|
||||
c = FakeConn([("x", 3)], ["k", "v"])
|
||||
r1 = await A.q_one("SELECT k, v FROM t WHERE id=%s", ("x",), conn=c)
|
||||
c2 = FakeConn([(9,)], ["max_seq"])
|
||||
r2 = await A.q_all("SELECT MAX(client_seq) AS max_seq FROM t WHERE tenant_id=%s",
|
||||
("T1",), conn=c2)
|
||||
c3 = FakeConn([], [])
|
||||
r3 = await A.q_one("SELECT 1 FROM t WHERE id=%s", ("none",), conn=c3)
|
||||
return r1, r2, r3
|
||||
|
||||
r1, r2, r3 = asyncio.get_event_loop().run_until_complete(await_paths())
|
||||
check("await q_one 得 dict", isinstance(r1, dict) and r1.get("v") == 3, str(r1))
|
||||
check("await q_all 得 list", isinstance(r2, list) and len(r2) == 1)
|
||||
check("await 空结果不抛（返回空 dict）", isinstance(r3, dict) and not r3)
|
||||
|
||||
# --- 4. q_one 无行语义：空 dict 而非 None ---
|
||||
print("\n[5] q_one 无行 → 空 dict（与 (row or {}).get / if row: 兼容）")
|
||||
empty = A.q_one("SELECT 1 FROM t WHERE id=%s", ("nope",), conn=FakeConn([], []))
|
||||
check("falsy（if existing: 走 else 分支）", not empty)
|
||||
check("(row or {}).get('x') 不炸", (empty or {}).get("x") is None)
|
||||
check("int((row or {}).get('max_seq') or 0) == 0",
|
||||
int((empty or {}).get("max_seq") or 0) == 0)
|
||||
|
||||
# --- 5. 异常不抛断链（fail-soft：返回空容器） ---
|
||||
print("\n[6] 连接异常时 q_all/q_one 返回空容器而非抛异常")
|
||||
|
||||
class BoomConn:
|
||||
def cursor(self):
|
||||
raise RuntimeError("connection lost")
|
||||
|
||||
try:
|
||||
rr = A.q_all("SELECT 1", (), conn=BoomConn())
|
||||
check("q_all 异常降级为空 list", isinstance(rr, list) and len(rr) == 0)
|
||||
ro = A.q_one("SELECT 1", (), conn=BoomConn())
|
||||
check("q_one 异常降级为空 dict", isinstance(ro, dict) and not ro)
|
||||
except Exception as exc:
|
||||
check("q_all/q_one 不应抛异常", False, repr(exc))
|
||||
|
||||
# --- 6. module_dbname：失败不抛、不返回硬编码库名 ---
|
||||
print("\n[7] module_dbname(module) 失败不抛")
|
||||
v = A.module_dbname("pbl_runtime_ext", default=None)
|
||||
check("返回 str 或 None（不抛异常）", v is None or isinstance(v, str), str(v))
|
||||
check("未解析时返回 default（不硬编码库名）",
|
||||
v is None or v == A.get_module_dbname("pbl_runtime_ext"))
|
||||
check("default 参数生效", A.module_dbname("__no_such_module__", default="D") in ("D",)
|
||||
or isinstance(A.module_dbname("__no_such_module__", default="D"), str))
|
||||
|
||||
# --- 7. db.get_sor_context 三级降级 ---
|
||||
print("\n[8] db.get_sor_context(env, module) 降级顺序")
|
||||
|
||||
class FakeEnv:
|
||||
def __init__(self, injected=False):
|
||||
self._injected = injected
|
||||
if injected:
|
||||
self.get_sor_context = lambda e, m: ("INJECTED", m)
|
||||
|
||||
check("env 注入 helper 时优先复用",
|
||||
D.get_sor_context(FakeEnv(True), "pbl_compiler") == ("INJECTED", "pbl_compiler"))
|
||||
try:
|
||||
D.get_sor_context(FakeEnv(False), None)
|
||||
check("无注入且无库名时应 fail-closed 抛 RuntimeError", False)
|
||||
except RuntimeError:
|
||||
check("无注入且无库名时 fail-closed 抛 RuntimeError", True)
|
||||
except Exception as exc:
|
||||
check("无注入且无库名时 fail-closed 抛 RuntimeError", False, repr(exc))
|
||||
|
||||
# --- 8. 下游模块导入闭包（原 Bug 的致命症状：包导入即 ImportError） ---
|
||||
print("\n[9] 下游模块导入闭包（pbl_runtime_ext / pbl_assessment / pbl_scense_ext）")
|
||||
for mod in ("pbl_runtime_ext", "pbl_runtime_ext.idempotency", "pbl_runtime_ext.rtx_db",
|
||||
"pbl_assessment", "pbl_scense_ext", "pbl_compiler.exporter"):
|
||||
try:
|
||||
__import__(mod)
|
||||
check("import %s 成功" % mod, True)
|
||||
except ImportError as exc:
|
||||
# 仅当失败原因是 pbl_common 契约符号缺失时才算本 Bug 复发
|
||||
bad = "pbl_common" in str(exc) and any(
|
||||
s in str(exc) for s in ("q_all", "q_one", "module_dbname",
|
||||
"get_sor_context", "crud"))
|
||||
check("import %s 无 pbl_common 契约符号缺失" % mod, not bad, str(exc))
|
||||
if not bad:
|
||||
print(" (基座依赖缺失导致的 ImportError，非本 Bug 范畴: %s)" % exc)
|
||||
except Exception as exc:
|
||||
check("import %s 无 pbl_common 契约符号缺失" % mod, True)
|
||||
print(" (非 ImportError: %r)" % (exc,))
|
||||
|
||||
print("\n" + "=" * 68)
|
||||
print("自测结果: PASS=%d FAIL=%d" % (len(PASS), len(FAIL)))
|
||||
if FAIL:
|
||||
print("失败项: %s" % ", ".join(FAIL))
|
||||
print("=" * 68)
|
||||
return 0 if not FAIL else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
321
scripts/verify_contract_symbols_all_modules.py
Normal file
321
scripts/verify_contract_symbols_all_modules.py
Normal file
@ -0,0 +1,321 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""跨模块契约符号闭包静态核验（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量排查）。
|
||||
|
||||
背景
|
||||
----
|
||||
原 Bug YyDumUuSK9R9Q_3Ai23AD 的 develop 侧根因 G：``pbl_common.api`` 缺少 ``crud``
|
||||
符号，导致下游 ``pbl_assessment`` / ``pbl_scense_ext`` 在包导入期硬失败
|
||||
（ImportError），应用进程根本起不来（healthz=000 / pbls.py 进程数=0）。
|
||||
|
||||
该缺陷的**同类问题**不止报出来的那一个：任何 ``from pbl_common.X import Y`` 而
|
||||
``Y`` 在 ``pbl_common`` 对应模块里并不存在的情况，都会造成同样的导入期硬失败。
|
||||
本脚本对全部 pbl_* 模块做静态 AST 闭包核验，不依赖运行时基座（基座未
|
||||
pip install -e、内网 pypi 不可达时依然可跑），因此可在开发机/CI 上作为
|
||||
提交前门禁使用。
|
||||
|
||||
检查项
|
||||
------
|
||||
1. 收集 ``modules/pbl_common/pbl_common/*.py`` 每个文件顶层可导出符号
|
||||
（def / async def / class / 赋值名 / __all__ 字面量 / from-import 再导出名）。
|
||||
2. 扫描全部 ``modules/pbl_*/`` 与 ``apps/pbls/`` 下 .py 的
|
||||
``from pbl_common[.sub] import a, b, c`` 语句，逐个符号核对是否存在。
|
||||
3. 额外核对 ``pbl_common/api.py`` 的 CONTRACT_SYMBOLS 与 __all__ 声明的符号
|
||||
是否真的在 api.py 命名空间内可解析（声明了但没实现 = 死契约）。
|
||||
4. 输出机器可读结论：PASS / FAIL + 明细，退出码 0/1，可直接接 CI。
|
||||
|
||||
用法::
|
||||
|
||||
python3 apps/pbls/scripts/verify_contract_symbols_all_modules.py
|
||||
python3 apps/pbls/scripts/verify_contract_symbols_all_modules.py --json out.json
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Dict, List, Set, Tuple
|
||||
|
||||
# ---------------------------------------------------------------- 路径常量
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
APP_ROOT = os.path.dirname(HERE) # apps/pbls
|
||||
WS_ROOT = os.path.dirname(os.path.dirname(APP_ROOT)) # 机构工作空间
|
||||
MODULES_DIR = os.path.join(WS_ROOT, "modules")
|
||||
COMMON_PKG = os.path.join(MODULES_DIR, "pbl_common", "pbl_common")
|
||||
|
||||
# 需要扫描 import 语句的根（模块 + 应用）
|
||||
SCAN_ROOTS: List[str] = [MODULES_DIR, APP_ROOT]
|
||||
|
||||
# 只关心 pbl_ 前缀的模块目录（避免扫到无关历史仓库）
|
||||
MODULE_PREFIX = "pbl_"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 工具函数
|
||||
def _iter_py_files(root: str):
|
||||
"""遍历目录下所有 .py 文件（跳过 venv/.git/__pycache__/wwwroot 生成物）。"""
|
||||
skip_dirs = {".git", "__pycache__", "venv", ".venv", "node_modules",
|
||||
"wwwroot", ".pytest_cache", "site-packages"}
|
||||
for dirpath, dirnames, filenames in os.walk(root):
|
||||
dirnames[:] = [d for d in dirnames if d not in skip_dirs]
|
||||
for fn in filenames:
|
||||
if fn.endswith(".py"):
|
||||
yield os.path.join(dirpath, fn)
|
||||
|
||||
|
||||
def _literal_names(node: ast.AST) -> Set[str]:
|
||||
"""从赋值右侧的 list/tuple/set 字面量里抽出字符串元素（用于 __all__）。"""
|
||||
out: Set[str] = set()
|
||||
if isinstance(node, (ast.List, ast.Tuple, ast.Set)):
|
||||
for elt in node.elts:
|
||||
if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
|
||||
out.add(elt.value)
|
||||
return out
|
||||
|
||||
|
||||
def collect_module_symbols(pkg_dir: str) -> Dict[str, Set[str]]:
|
||||
"""收集 pbl_common 包内每个子模块的顶层可导出符号。
|
||||
|
||||
返回 {子模块名(不含 .py): 符号集合}，另外附加一个聚合键 ``"*"`` 表示
|
||||
``from pbl_common import X`` 时包 __init__ 可见的符号。
|
||||
"""
|
||||
table: Dict[str, Set[str]] = {}
|
||||
for path in sorted(_iter_py_files(pkg_dir)):
|
||||
rel = os.path.relpath(path, pkg_dir)
|
||||
if os.sep in rel: # 只取包顶层模块（pbl_common/xxx.py）
|
||||
continue
|
||||
mod = rel[:-3]
|
||||
try:
|
||||
tree = ast.parse(open(path, encoding="utf-8").read(), filename=path)
|
||||
except SyntaxError as exc: # 语法错本身就是缺陷，记为特殊符号
|
||||
table[mod] = {"__SYNTAX_ERROR__: %s" % exc}
|
||||
continue
|
||||
syms: Set[str] = set()
|
||||
for node in tree.body:
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||||
syms.add(node.name)
|
||||
elif isinstance(node, ast.Assign):
|
||||
for tgt in node.targets:
|
||||
if isinstance(tgt, ast.Name):
|
||||
syms.add(tgt.id)
|
||||
# __all__ = [...] 里的名字也算「本模块对外承诺的符号」
|
||||
if tgt.id in ("__all__", "CONTRACT_SYMBOLS"):
|
||||
syms |= _literal_names(node.value)
|
||||
elif isinstance(tgt, ast.Tuple):
|
||||
for e in tgt.elts:
|
||||
if isinstance(e, ast.Name):
|
||||
syms.add(e.id)
|
||||
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
|
||||
syms.add(node.target.id)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
# from x import y → y 在本模块命名空间可见（再导出）
|
||||
for al in node.names:
|
||||
syms.add(al.asname or al.name)
|
||||
elif isinstance(node, ast.Import):
|
||||
for al in node.names:
|
||||
syms.add((al.asname or al.name).split(".")[0])
|
||||
elif isinstance(node, (ast.If, ast.Try)):
|
||||
# 条件/异常分支里的顶层定义同样计入（兼容 shim 常用 try-import）
|
||||
for sub in ast.walk(node):
|
||||
if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||||
syms.add(sub.name)
|
||||
elif isinstance(sub, ast.Assign):
|
||||
for tgt in sub.targets:
|
||||
if isinstance(tgt, ast.Name):
|
||||
syms.add(tgt.id)
|
||||
elif isinstance(sub, ast.ImportFrom):
|
||||
for al in sub.names:
|
||||
syms.add(al.asname or al.name)
|
||||
table[mod] = syms
|
||||
|
||||
# 包级聚合：__init__.py 的符号 + 所有子模块名本身（from pbl_common import crud 也合法）
|
||||
agg: Set[str] = set(table.get("__init__", set()))
|
||||
agg |= set(table.keys())
|
||||
for syms in table.values():
|
||||
agg |= syms
|
||||
table["*"] = agg
|
||||
return table
|
||||
|
||||
|
||||
PROVIDER_PREFIXES = ("pbl_",)
|
||||
|
||||
|
||||
def collect_imports(path: str) -> List[Tuple[int, str, List[str]]]:
|
||||
"""抽出文件里所有 from pbl_*[...] import ... 语句（跨模块契约引用）。
|
||||
|
||||
返回 [(lineno, 模块路径, [符号...]), ...]
|
||||
"""
|
||||
try:
|
||||
src = open(path, encoding="utf-8").read()
|
||||
except (OSError, UnicodeDecodeError):
|
||||
return []
|
||||
try:
|
||||
tree = ast.parse(src, filename=path)
|
||||
except SyntaxError:
|
||||
return []
|
||||
out: List[Tuple[int, str, List[str]]] = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ImportFrom) and node.module and \
|
||||
any(node.module == p or node.module.startswith(p + ".")
|
||||
for p in PROVIDER_PREFIXES):
|
||||
names = [al.name for al in node.names if al.name != "*"]
|
||||
if names:
|
||||
out.append((node.lineno, node.module, names))
|
||||
return out
|
||||
|
||||
|
||||
def scan_targets() -> List[str]:
|
||||
"""待扫描文件清单：所有 pbl_* 模块 + apps/pbls 下的 .py。"""
|
||||
files: List[str] = []
|
||||
for root in SCAN_ROOTS:
|
||||
if not os.path.isdir(root):
|
||||
continue
|
||||
if root == MODULES_DIR:
|
||||
for name in sorted(os.listdir(root)):
|
||||
if not name.startswith(MODULE_PREFIX):
|
||||
continue
|
||||
mod_dir = os.path.join(root, name)
|
||||
if os.path.isdir(mod_dir):
|
||||
files.extend(_iter_py_files(mod_dir))
|
||||
else:
|
||||
files.extend(_iter_py_files(root))
|
||||
return sorted(set(files))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 主核验
|
||||
def verify() -> Dict:
|
||||
if not os.path.isdir(COMMON_PKG):
|
||||
return {"ok": False, "fatal": "pbl_common 包目录不存在: %s" % COMMON_PKG,
|
||||
"checked_files": 0, "violations": [], "dead_contract": []}
|
||||
|
||||
symtab = collect_module_symbols(COMMON_PKG)
|
||||
# 全部 pbl_* 模块包的符号表：{包名: {子模块名: 符号集}}
|
||||
pkg_tabs: Dict[str, Dict[str, Set[str]]] = {"pbl_common": symtab}
|
||||
for name in sorted(os.listdir(MODULES_DIR)):
|
||||
if not name.startswith(MODULE_PREFIX) or name == "pbl_common":
|
||||
continue
|
||||
pkg_dir = os.path.join(MODULES_DIR, name, name)
|
||||
if os.path.isdir(pkg_dir):
|
||||
pkg_tabs[name] = collect_module_symbols(pkg_dir)
|
||||
files = scan_targets()
|
||||
|
||||
violations: List[Dict] = []
|
||||
for path in files:
|
||||
# provider 包自身内部互引不算跨模块契约违规（内部结构可自由重构）
|
||||
if any(os.path.abspath(path).startswith(
|
||||
os.path.abspath(os.path.join(MODULES_DIR, k, k)) + os.sep)
|
||||
for k in pkg_tabs):
|
||||
continue
|
||||
own_pkg = os.path.relpath(path, MODULES_DIR).split(os.sep)[0]
|
||||
for lineno, mod, names in collect_imports(path):
|
||||
provider = mod.split(".", 0)[0].split(".")[0]
|
||||
tab = pkg_tabs.get(provider)
|
||||
if tab is None:
|
||||
continue # 非本仓 pbl_* 模块（如平台包），跳过
|
||||
if provider == own_pkg:
|
||||
continue # 包内自引不算跨模块契约
|
||||
if "." not in mod:
|
||||
avail = tab.get("*", set())
|
||||
sub = "%s(包)" % provider
|
||||
else:
|
||||
submod = mod.split(".", 1)[1]
|
||||
avail = tab.get(submod, set())
|
||||
sub = "%s.%s" % (provider, submod)
|
||||
if submod not in tab:
|
||||
violations.append({
|
||||
"file": os.path.relpath(path, WS_ROOT),
|
||||
"line": lineno,
|
||||
"module": mod,
|
||||
"symbol": "<module>",
|
||||
"reason": "子模块 %s/%s.py 不存在" % (provider, submod),
|
||||
})
|
||||
continue
|
||||
for nm in names:
|
||||
if nm not in avail:
|
||||
violations.append({
|
||||
"file": os.path.relpath(path, WS_ROOT),
|
||||
"line": lineno,
|
||||
"module": sub,
|
||||
"symbol": nm,
|
||||
"reason": "符号未在 %s 顶层定义/再导出（导入期 ImportError 风险）" % sub,
|
||||
})
|
||||
|
||||
# 死契约核验：api.py 里 CONTRACT_SYMBOLS / __all__ 声明了但命名空间没有
|
||||
dead: List[Dict] = []
|
||||
api_path = os.path.join(COMMON_PKG, "api.py")
|
||||
if os.path.isfile(api_path):
|
||||
src = open(api_path, encoding="utf-8").read()
|
||||
tree = ast.parse(src, filename=api_path)
|
||||
declared: Dict[str, Set[str]] = {}
|
||||
defined: Set[str] = set()
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.Assign):
|
||||
for tgt in node.targets:
|
||||
if isinstance(tgt, ast.Name) and tgt.id in ("__all__", "CONTRACT_SYMBOLS"):
|
||||
declared[tgt.id] = _literal_names(node.value)
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||||
defined.add(node.name)
|
||||
elif isinstance(node, ast.Assign):
|
||||
for tgt in node.targets:
|
||||
if isinstance(tgt, ast.Name):
|
||||
defined.add(tgt.id)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
for al in node.names:
|
||||
defined.add(al.asname or al.name)
|
||||
for key, names in declared.items():
|
||||
for nm in sorted(names):
|
||||
if nm not in defined:
|
||||
dead.append({"list": key, "symbol": nm,
|
||||
"reason": "声明于 %s 但 api.py 命名空间不可解析" % key})
|
||||
|
||||
return {
|
||||
"ok": not violations and not dead,
|
||||
"checked_files": len(files),
|
||||
"common_modules": sorted(k for k in symtab if k != "*"),
|
||||
"provider_packages": sorted(pkg_tabs.keys()),
|
||||
"crud_symbol_present": "crud" in symtab.get("api", set()),
|
||||
"violations": violations,
|
||||
"dead_contract": dead,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="pbl_common 契约符号跨模块闭包核验")
|
||||
ap.add_argument("--json", dest="json_out", default=None,
|
||||
help="把结果写入指定 JSON 文件")
|
||||
args = ap.parse_args()
|
||||
|
||||
res = verify()
|
||||
print("=" * 72)
|
||||
print("pbl_common 契约符号闭包核验（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量排查）")
|
||||
print("=" * 72)
|
||||
if res.get("fatal"):
|
||||
print("FATAL: %s" % res["fatal"])
|
||||
return 2
|
||||
print("扫描 .py 文件数 : %d" % res["checked_files"])
|
||||
print("pbl_common 子模块数 : %d" % len(res["common_modules"]))
|
||||
print("核验 provider 包数 : %d (%s)" % (len(res["provider_packages"]),
|
||||
", ".join(res["provider_packages"])))
|
||||
print("api.crud 符号在位 : %s" % res["crud_symbol_present"])
|
||||
print("跨模块导入符号违规 : %d" % len(res["violations"]))
|
||||
print("死契约(声明未实现) : %d" % len(res["dead_contract"]))
|
||||
for v in res["violations"][:50]:
|
||||
print(" [VIOLATION] %s:%s from %s import %s → %s"
|
||||
% (v["file"], v["line"], v["module"], v["symbol"], v["reason"]))
|
||||
for d in res["dead_contract"][:50]:
|
||||
print(" [DEAD] %s 声明 '%s' → %s" % (d["list"], d["symbol"], d["reason"]))
|
||||
print("-" * 72)
|
||||
print("结论: %s" % ("PASS —— 全部 pbl_* 模块间跨模块导入符号均可解析"
|
||||
if res["ok"] else "FAIL —— 存在导入期硬失败风险，须修复后再部署"))
|
||||
|
||||
if args.json_out:
|
||||
with open(args.json_out, "w", encoding="utf-8") as fh:
|
||||
json.dump(res, fh, ensure_ascii=False, indent=2)
|
||||
print("结果已写入: %s" % args.json_out)
|
||||
return 0 if res["ok"] else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
x
Reference in New Issue
Block a user