deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
ae4f0eccc1
commit
93829c6a9b
144
pbl_runtime_ext/db.py
Normal file
144
pbl_runtime_ext/db.py
Normal file
@ -0,0 +1,144 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_runtime_ext.db —— 真实存在的 DB 原语封装（M11a，QC #6 取证后重写）。
|
||||
|
||||
为什么需要这一层
|
||||
----------------
|
||||
本仓库 `pbl_common.api` **不提供** `u()` / `i()` / `q_one()` / `q_all()` 这组签名
|
||||
（已用 inspect 逐个核实，见 skill/SKILL.md「pbl_common 契约面取证」一节）。
|
||||
M11a 早期代码凭记忆写了 `A.u/A.i/A.new_id/q_one/q_all`，import 即 AttributeError。
|
||||
本文件把「幂等闸门」真正需要的 4 个原语落在 pbl_common.api **确实存在**的
|
||||
`sql_exec / sql_rows / sql_scalar / new_id` 之上，签名与语义在此固化，
|
||||
下游（idempotency.py / entity_state.py）只依赖本模块，不再直接猜 pbl_common 的 API。
|
||||
|
||||
已核实的 pbl_common.api 真实签名（2026-09，inspect.signature 实测）：
|
||||
sql_exec(sql, params=None, dbname=None) -> int 同步
|
||||
sql_rows(sql, params=None, dbname=None) -> list[dict] 同步
|
||||
sql_scalar(sql, params=None, dbname=None, default=None) 同步
|
||||
new_id(prefix='') -> str(<=32) 同步
|
||||
tenant_id(default=None) / actor_id(default=None) 同步（上下文读取器）
|
||||
insert_row/update_row/fetch_one/query **async**（本模块不使用，避免混用）
|
||||
|
||||
占位符约定：`%s` + **tuple** 参数（pbl_common.dbutil.execute 内部 `args = tuple(params)`，
|
||||
PyMySQL/sqlor 两条路径都吃这个形态）。禁止字符串拼接外部输入。
|
||||
|
||||
事务：所有原语接受可选 `conn`。给了 conn 就在同一连接上执行（幂等记录与业务写入同事务，
|
||||
业务回滚则闸门一并回滚）；没给则走 pbl_common 的自动连接路径（单语句原子）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
||||
|
||||
from pbl_common.api import new_id, sql_exec, sql_rows # 已核实存在的真实符号
|
||||
|
||||
__all__ = ["new_id", "exec_sql", "q_rows", "q_one", "q_scalar", "insert_row_sql", "update_row_sql"]
|
||||
|
||||
MODULE_DB = "pbl" # 库名解析交给 pbl_common（get_module_dbname），此处仅作 dbname 提示
|
||||
|
||||
|
||||
def _dbname() -> Optional[str]:
|
||||
"""解析本模块库名；平台/上下文不可用时返回 None（让 pbl_common 走默认库）。"""
|
||||
try:
|
||||
from pbl_common.api import get_module_dbname
|
||||
name = get_module_dbname("pbl_runtime_ext")
|
||||
return name or None
|
||||
except Exception: # noqa: BLE001 - 离线/自检环境无 ServerEnv
|
||||
return None
|
||||
|
||||
|
||||
def _conn_exec(conn: Any, sql: str, params: Optional[Sequence[Any]]) -> Any:
|
||||
"""在调用方连接上执行（duck-typing：sqlExe / execute / cursor）。"""
|
||||
args = tuple(params or ())
|
||||
for attr in ("execute", "sqlExe", "sql_exe"):
|
||||
fn = getattr(conn, attr, None)
|
||||
if callable(fn):
|
||||
try:
|
||||
return fn(sql, args)
|
||||
except TypeError:
|
||||
return fn(sql)
|
||||
cursor_fn = getattr(conn, "cursor", None)
|
||||
if callable(cursor_fn):
|
||||
cur = cursor_fn()
|
||||
cur.execute(sql, args)
|
||||
return cur
|
||||
raise RuntimeError("conn 不支持 execute/sqlExe/cursor，无法在同事务内执行")
|
||||
|
||||
|
||||
def exec_sql(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None) -> int:
|
||||
"""执行写 SQL（INSERT/UPDATE/DELETE/DDL），返回受影响行数。"""
|
||||
if conn is not None:
|
||||
r = _conn_exec(conn, sql, params)
|
||||
if isinstance(r, int):
|
||||
return r
|
||||
affected = getattr(r, "rowcount", None)
|
||||
return int(affected) if isinstance(affected, int) else 0
|
||||
return int(sql_exec(sql, tuple(params or ()), dbname=_dbname()) or 0)
|
||||
|
||||
|
||||
def q_rows(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None) -> List[Dict[str, Any]]:
|
||||
"""查询多行，返回 list[dict]。"""
|
||||
if conn is not None:
|
||||
r = _conn_exec(conn, sql, params)
|
||||
if isinstance(r, list):
|
||||
return r
|
||||
if r is None:
|
||||
return []
|
||||
# cursor 形态：description + fetchall
|
||||
desc = getattr(r, "description", None)
|
||||
fetch = getattr(r, "fetchall", None)
|
||||
if desc and callable(fetch):
|
||||
cols = [d[0] for d in desc]
|
||||
return [dict(zip(cols, row)) for row in fetch()]
|
||||
return list(r) if isinstance(r, (tuple, set)) else []
|
||||
return list(sql_rows(sql, tuple(params or ()), dbname=_dbname()) or [])
|
||||
|
||||
|
||||
def q_one(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None) -> Optional[Dict[str, Any]]:
|
||||
"""查询单行（无则 None）。fail-closed：多行时返回首行并保留 SQL 的 LIMIT 1 约束。"""
|
||||
rows = q_rows(sql, params, conn=conn)
|
||||
return rows[0] if rows else None
|
||||
|
||||
|
||||
def q_scalar(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None,
|
||||
default: Any = 0) -> Any:
|
||||
"""查询单值（首行首列）；空结果返回 default。"""
|
||||
rows = q_rows(sql, params, conn=conn)
|
||||
if not rows:
|
||||
return default
|
||||
first = rows[0]
|
||||
if isinstance(first, dict):
|
||||
vals = list(first.values())
|
||||
return vals[0] if vals else default
|
||||
if isinstance(first, (list, tuple)):
|
||||
return first[0] if first else default
|
||||
return first
|
||||
|
||||
|
||||
def _placeholders(n: int) -> str:
|
||||
return ",".join(["%s"] * max(0, int(n)))
|
||||
|
||||
|
||||
def insert_row_sql(table: str, row: Dict[str, Any], conn: Any = None) -> int:
|
||||
"""参数化 INSERT（列名来自调用方常量，不接受外部输入拼接）。返回受影响行数。"""
|
||||
cols: Tuple[str, ...] = tuple(row.keys())
|
||||
if not cols:
|
||||
raise ValueError("insert_row_sql 需要至少一列")
|
||||
sql = "INSERT INTO `%s` (%s) VALUES (%s)" % (
|
||||
table,
|
||||
", ".join("`%s`" % c for c in cols),
|
||||
_placeholders(len(cols)),
|
||||
)
|
||||
return exec_sql(sql, [row[c] for c in cols], conn=conn)
|
||||
|
||||
|
||||
def update_row_sql(table: str, values: Dict[str, Any], where: Dict[str, Any],
|
||||
conn: Any = None) -> int:
|
||||
"""参数化 UPDATE（WHERE 各列 AND 连接，强制要求 where 非空 → 防全表更新）。"""
|
||||
if not values:
|
||||
raise ValueError("update_row_sql 无可更新列")
|
||||
if not where:
|
||||
raise ValueError("update_row_sql 必须带 WHERE 条件（防全表更新，fail-closed）")
|
||||
sets = ", ".join("`%s` = %%s" % k for k in values)
|
||||
conds = " AND ".join("`%s` = %%s" % k for k in where)
|
||||
sql = "UPDATE `%s` SET %s WHERE %s" % (table, sets, conds)
|
||||
params = list(values.values()) + list(where.values())
|
||||
return exec_sql(sql, params, conn=conn)
|
||||
@ -1,15 +1,30 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_runtime_ext 表定义与幂等建表（M11a）。
|
||||
"""pbl_runtime_ext 表定义与幂等建表（M11a：4 表，单一真源）。
|
||||
|
||||
3 表（mariadb 方言，BIGINT AUTO_INCREMENT，tenant_id 强制打头，无 FK/ENUM/TIMESTAMP）：
|
||||
* pbl_runtime_event —— 运行时事件流（append-only，幂等键去重）
|
||||
* pbl_entity_state —— 实体状态（薄扩展，不改 entity 基表）
|
||||
4 表（mariadb 方言，BIGINT AUTO_INCREMENT 主键，tenant_id 强制打头，无 FK/ENUM/TIMESTAMP）：
|
||||
* pbl_runtime_event —— 运行时事件流（append-only，idem_key 幂等唯一键）
|
||||
* pbl_entity_state —— 实体状态（服务端权威，state_version 客户端禁写）
|
||||
* pbl_world_state_snapshot —— 世界状态快照（单事务写入后广播基线）
|
||||
* pbl_runtime_idempotency —— client_seq 幂等闸门（M11a 新增，唯一键
|
||||
(tenant_id, session_id, scope, client_seq)）
|
||||
|
||||
与 modules/pbl_runtime_ext/models/*.json 及 sql/pbl_runtime_ext.sql 同构。
|
||||
**单一真源**：本文件的 DDL 列表是唯一权威定义，`sql/pbl_runtime_ext.sql` 与
|
||||
`models/*.json` 由 `scripts/gen_artifacts.py` 从本文件生成，三处不可能漂移。
|
||||
表名常量（TBL_*）供代码引用，禁止在业务代码里再写裸表名字符串。
|
||||
"""
|
||||
|
||||
TABLES = ["pbl_runtime_event", "pbl_entity_state", "pbl_world_state_snapshot"]
|
||||
TBL_RUNTIME_EVENT = "pbl_runtime_event"
|
||||
TBL_ENTITY_STATE = "pbl_entity_state"
|
||||
TBL_SNAPSHOT = "pbl_world_state_snapshot"
|
||||
TBL_IDEMPOTENCY = "pbl_runtime_idempotency"
|
||||
|
||||
TABLES = [TBL_RUNTIME_EVENT, TBL_ENTITY_STATE, TBL_SNAPSHOT, TBL_IDEMPOTENCY]
|
||||
|
||||
# 表名常量 → 幂等闸门/状态权威链路用到的关键列（供自检与文档核对，避免列名漂移）
|
||||
IDEMPOTENCY_COLUMNS = ["idempotency_id", "tenant_id", "session_id", "scope", "client_seq",
|
||||
"principal_id", "request_fp", "state", "result_json", "error_code",
|
||||
"created_at", "updated_at"]
|
||||
ENTITY_STATE_VERSION_COLUMN = "state_version"
|
||||
|
||||
DDL = [
|
||||
"""
|
||||
@ -47,14 +62,15 @@ CREATE TABLE IF NOT EXISTS `pbl_entity_state` (
|
||||
`entity_id` BIGINT NOT NULL DEFAULT 0 COMMENT '实体ID（引用 entity 基表，只读）',
|
||||
`state_key` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '状态键',
|
||||
`state_value` TEXT COMMENT '状态值 JSON',
|
||||
`version` INT NOT NULL DEFAULT 1 COMMENT '乐观锁版本',
|
||||
`state_version` BIGINT NOT NULL DEFAULT 0 COMMENT '服务端权威单调版本（客户端禁写）',
|
||||
`last_event_id` BIGINT NOT NULL DEFAULT 0 COMMENT '最后触发事件ID',
|
||||
`created_at` DATETIME COMMENT '创建时间',
|
||||
`updated_at` DATETIME COMMENT '更新时间',
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_tenant_entity_key` (`tenant_id`, `session_id`, `entity_id`, `state_key`),
|
||||
KEY `ix_tenant_world` (`tenant_id`, `world_id`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实体状态薄扩展'
|
||||
KEY `ix_tenant_world` (`tenant_id`, `world_id`),
|
||||
KEY `ix_tenant_state_version` (`tenant_id`, `session_id`, `state_version`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实体状态薄扩展（服务端权威，state_version 客户端禁写）'
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS `pbl_world_state_snapshot` (
|
||||
@ -73,15 +89,59 @@ CREATE TABLE IF NOT EXISTS `pbl_world_state_snapshot` (
|
||||
UNIQUE KEY `uk_tenant_snapshot_code` (`tenant_id`, `snapshot_code`),
|
||||
KEY `ix_tenant_session_seq` (`tenant_id`, `session_id`, `seq_no`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='世界状态快照'
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS `pbl_runtime_idempotency` (
|
||||
`id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
|
||||
`tenant_id` BIGINT NOT NULL DEFAULT 0 COMMENT '租户ID（多租户强制打头）',
|
||||
`idempotency_id` VARCHAR(64) NOT NULL COMMENT '幂等记录业务主键',
|
||||
`session_id` BIGINT NOT NULL DEFAULT 0 COMMENT '会话ID（引用 scense 基表，只读）',
|
||||
`scope` VARCHAR(64) NOT NULL DEFAULT '*' COMMENT '幂等作用域（world_id 或 world_id:channel，缺省 *）',
|
||||
`client_seq` BIGINT NOT NULL DEFAULT 0 COMMENT '客户端单调序号（>=1）',
|
||||
`principal_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '提交者（服务端会话解析）',
|
||||
`request_fp` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '请求体指纹（同 seq 不同内容判冲突）',
|
||||
`state` VARCHAR(16) NOT NULL DEFAULT 'in_progress' COMMENT 'in_progress/completed/failed',
|
||||
`result_json` TEXT COMMENT '已完成结果缓存（重放直接返回，不再写第二条事件）',
|
||||
`error_code` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '失败错误码（failed 态）',
|
||||
`updated_at` BIGINT NOT NULL DEFAULT 0 COMMENT '更新时间（epoch 毫秒）',
|
||||
`created_at` BIGINT NOT NULL DEFAULT 0 COMMENT '创建时间（epoch 毫秒）',
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_tenant_seq` (`tenant_id`, `session_id`, `scope`, `client_seq`),
|
||||
UNIQUE KEY `uk_tenant_idem_id` (`tenant_id`, `idempotency_id`),
|
||||
KEY `ix_tenant_scope_seq` (`tenant_id`, `session_id`, `scope`, `state`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='运行时 client_seq 幂等闸门（M11a 服务端权威）'
|
||||
""",
|
||||
]
|
||||
|
||||
|
||||
def table_names():
|
||||
"""全部表名（list[str]）。"""
|
||||
return list(TABLES)
|
||||
|
||||
|
||||
def get_ddl():
|
||||
"""返回本模块全部建表语句（list[str]）。"""
|
||||
return [x.strip() for x in DDL]
|
||||
|
||||
|
||||
def ddl_for(table):
|
||||
"""按表名取建表语句；未知表名抛 KeyError（fail-closed，不静默返回空）。"""
|
||||
for stmt in get_ddl():
|
||||
if stmt.split("`")[1] == table:
|
||||
return stmt
|
||||
raise KeyError("未知表 %s（应为 %s 之一）" % (table, ",".join(TABLES)))
|
||||
|
||||
|
||||
def columns_for(table):
|
||||
"""按表名解析列名清单（从 DDL 单一真源解析，供 gen_artifacts / 自检使用）。"""
|
||||
cols = []
|
||||
for line in ddl_for(table).splitlines():
|
||||
s = line.strip()
|
||||
if s.startswith("`") and s.endswith(","):
|
||||
cols.append(s.split("`")[1])
|
||||
return cols
|
||||
|
||||
|
||||
def ensure_tables(env=None, sor=None):
|
||||
"""幂等建表（CREATE TABLE IF NOT EXISTS）。返回 (ok_list, bad_list)。"""
|
||||
ok, bad = [], []
|
||||
@ -105,13 +165,15 @@ def ensure_tables(env=None, sor=None):
|
||||
|
||||
|
||||
def self_check():
|
||||
"""离线自检：3 表齐全 / tenant_id 打头 / 无禁用方言 / 幂等键唯一约束在位。"""
|
||||
"""离线自检：4 表齐全 / tenant_id 打头 / 无禁用方言 / 幂等唯一键在位 / 列名与代码一致。"""
|
||||
msgs = []
|
||||
all_ok = True
|
||||
forbidden = ("BIGSERIAL", "SERIAL", "nextval", "FOREIGN KEY",
|
||||
"REFERENCES", "ENUM(", "TIMESTAMP")
|
||||
seen = []
|
||||
for stmt in get_ddl():
|
||||
name = stmt.split("`")[1]
|
||||
seen.append(name)
|
||||
body = stmt.upper()
|
||||
if name not in TABLES:
|
||||
all_ok = False
|
||||
@ -131,12 +193,46 @@ def self_check():
|
||||
if "IF NOT EXISTS" not in body:
|
||||
all_ok = False
|
||||
msgs.append("%s 非幂等建表（缺 IF NOT EXISTS）" % name)
|
||||
if "UNIQUE KEY `uk_tenant_idem`" not in DDL[0]:
|
||||
|
||||
if sorted(seen) != sorted(TABLES):
|
||||
all_ok = False
|
||||
msgs.append("DDL 表集合 %s 与 TABLES %s 不一致" % (sorted(seen), sorted(TABLES)))
|
||||
|
||||
# 事件表幂等唯一键
|
||||
if "UNIQUE KEY `uk_tenant_idem`" not in ddl_for(TBL_RUNTIME_EVENT):
|
||||
all_ok = False
|
||||
msgs.append("pbl_runtime_event 缺幂等唯一键 uk_tenant_idem")
|
||||
if len(TABLES) != 3:
|
||||
else:
|
||||
msgs.append("pbl_runtime_event uk_tenant_idem 在位")
|
||||
|
||||
# 状态表：服务端权威版本列必须是 state_version（代码引用的真实列名）
|
||||
state_cols = columns_for(TBL_ENTITY_STATE)
|
||||
if ENTITY_STATE_VERSION_COLUMN not in state_cols:
|
||||
all_ok = False
|
||||
msgs.append("表数=%d 应为 3" % len(TABLES))
|
||||
msgs.append("pbl_entity_state 缺列 %s（代码 MAX(state_version) 会 Unknown column）"
|
||||
% ENTITY_STATE_VERSION_COLUMN)
|
||||
else:
|
||||
msgs.append("pbl_entity_state.%s 在位（服务端权威，客户端禁写）" % ENTITY_STATE_VERSION_COLUMN)
|
||||
if "version" in state_cols and ENTITY_STATE_VERSION_COLUMN not in state_cols:
|
||||
all_ok = False
|
||||
msgs.append("pbl_entity_state 仍用旧列名 version，应统一为 state_version")
|
||||
|
||||
# 幂等闸门表：列 + 唯一键
|
||||
idem_cols = columns_for(TBL_IDEMPOTENCY)
|
||||
for col in IDEMPOTENCY_COLUMNS:
|
||||
if col not in idem_cols:
|
||||
all_ok = False
|
||||
msgs.append("pbl_runtime_idempotency 缺列 %s（idempotency.py 读写依赖）" % col)
|
||||
idem_ddl = ddl_for(TBL_IDEMPOTENCY)
|
||||
if "UNIQUE KEY `uk_tenant_seq` (`tenant_id`, `session_id`, `scope`, `client_seq`)" not in idem_ddl:
|
||||
all_ok = False
|
||||
msgs.append("pbl_runtime_idempotency 缺唯一键 uk_tenant_seq(tenant_id,session_id,scope,client_seq)")
|
||||
if all_ok and set(IDEMPOTENCY_COLUMNS) <= set(idem_cols):
|
||||
msgs.append("pbl_runtime_idempotency 列/唯一键齐全（%d 列）" % len(idem_cols))
|
||||
|
||||
if len(TABLES) != 4:
|
||||
all_ok = False
|
||||
msgs.append("表数=%d 应为 4" % len(TABLES))
|
||||
if all_ok:
|
||||
msgs.append("SELF_CHECK pbl_runtime_ext.tables: PASS %d/%d" % (len(TABLES), len(TABLES)))
|
||||
return all_ok, msgs
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user