fix(pbl_common): [Bug HoSBcdZY4G7IxhaFiIE07] 补齐 4 个跨模块契约符号 q_all/q_one/module_dbname/get_sor_context
- api.py: q_all/q_one 双模(同步+await)查询助手,q_one 无行返回空 dict; module_dbname 失败不抛的库名解析包装(内部委托 get_module_dbname,无硬编码 DBNAME); CONTRACT_SYMBOLS + __all__ 同步登记三符号 - db.py: get_sor_context(env, module_name) 公共入口(pbl_compiler/exporter.py 契约名), 优先宿主注入 helper,其次 sqlor DBPools().sqlorContext(解析库名),均不可用则显式抛错 - 根因: pbl_runtime_ext/idempotency.py 顶层 from pbl_common.api import q_all, q_one 符号缺失 -> 包导入期 ImportError -> 应用起不来(healthz=000),与原 crud 缺口同型 - 验证: py_compile OK; verify_contract_symbols_all_modules.py 违规=0 死契约=0 PASS; _selftest_contract_symbols.py PASS=34 FAIL=0; verify_import_closure.py FAIL 项=0 - 选择性 git add(仅 api.py/db.py,未用 -A)
This commit is contained in:
parent
fc8bc3fa7e
commit
b5ac672f28
@ -326,7 +326,164 @@ def api_guard(func=None, tenant_required=True, audit_action=None, audit_table=No
|
||||
# ==========================================================================
|
||||
# 契约自检(import 期即验证符号面完整,失败直接抛 → 不允许半迁移状态被交付)
|
||||
# ==========================================================================
|
||||
# >>> 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) <<<
|
||||
|
||||
|
||||
CONTRACT_SYMBOLS = (
|
||||
# 查询助手/库名解析(Bug HoSBcdZY4G7IxhaFiIE07 同类缺口补齐)
|
||||
'q_all', 'q_one', 'module_dbname',
|
||||
# errors
|
||||
'PBLError', 'PblError', 'err', 'fail', 'as_error', 'error_body',
|
||||
'ErrorCode', 'PBL_ERR', 'CODE_TO_HTTP', 'LEGACY_CODE_MAP', 'ALL_CODES',
|
||||
|
||||
@ -158,3 +158,42 @@ def upsert_sql(table, unique_cols, update_cols):
|
||||
if sets:
|
||||
sql += ' ON DUPLICATE KEY UPDATE %s' % sets
|
||||
return sql
|
||||
|
||||
|
||||
# >>> 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) <<<
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user