diff --git a/pbl_common/api.py b/pbl_common/api.py index 00c4737..db24cf0 100644 --- a/pbl_common/api.py +++ b/pbl_common/api.py @@ -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', diff --git a/pbl_common/db.py b/pbl_common/db.py index 0b951a6..8cb71b7 100644 --- a/pbl_common/db.py +++ b/pbl_common/db.py @@ -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) <<<