147 lines
6.0 KiB
Python
147 lines
6.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
pbl_common.tables —— 公共表 DDL（mariadb 方言）
|
||
|
||
契约（docs/01-design/data-model.md + projects/pbls/env/test.json ddl 段）：
|
||
- 方言 mariadb：主键 `id BIGINT NOT NULL AUTO_INCREMENT`
|
||
- 禁止 FOREIGN KEY / REFERENCES / ENUM / TIMESTAMP
|
||
- 时间列一律 DATETIME
|
||
- 每表首列必须 tenant_id varchar(64) NOT NULL，且索引以 tenant_id 打头
|
||
- 编码列 varchar(32)
|
||
- ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
|
||
本模块只声明 pbl_common 自有表（审计 + 幂等种子记录）；
|
||
业务表 DDL 在各模块 tables.py 与 apps/pbls/scripts/ddl/pbls_tables.sql。
|
||
"""
|
||
|
||
from pbl_common.errors import DbError
|
||
|
||
DIALECT = 'mariadb'
|
||
|
||
# 禁用元素（ensure_tables 前自检，命中即抛错，防方言漂移）
|
||
FORBIDDEN_TOKENS = ('FOREIGN KEY', 'REFERENCES ', 'ENUM(', 'TIMESTAMP', 'BIGSERIAL', 'SERIAL', 'nextval')
|
||
|
||
TABLES = {
|
||
'pbl_audit_log': {
|
||
'comment': 'PBL 审计日志（append-only，审计独立性）',
|
||
'columns': [
|
||
('tenant_id', "varchar(64) NOT NULL COMMENT '租户ID（强制打头）'"),
|
||
('action', "varchar(32) NOT NULL COMMENT '动作（白名单 AUDIT_ACTIONS）'"),
|
||
('action_raw', "varchar(64) DEFAULT NULL COMMENT '原始动作名（未登记动作留痕）'"),
|
||
('resource_type', "varchar(64) DEFAULT NULL COMMENT '资源类型（表名）'"),
|
||
('resource_id', "varchar(64) DEFAULT NULL COMMENT '资源主键'"),
|
||
('user_id', "varchar(64) DEFAULT NULL COMMENT '操作人'"),
|
||
('role', "varchar(64) DEFAULT NULL COMMENT '操作人角色'"),
|
||
('session_id', "varchar(128) DEFAULT NULL COMMENT '会话ID'"),
|
||
('trace_id', "varchar(64) DEFAULT NULL COMMENT '调用链追踪ID'"),
|
||
('result', "varchar(16) NOT NULL DEFAULT 'success' COMMENT '结果 success/fail/deny'"),
|
||
('detail', "longtext DEFAULT NULL COMMENT '明细 JSON'"),
|
||
('created_at', "datetime NOT NULL COMMENT '创建时间'"),
|
||
],
|
||
'indexes': [
|
||
('PRIMARY KEY', '(`id`)'),
|
||
('KEY `idx_audit_tenant_time`', '(`tenant_id`,`created_at`)'),
|
||
('KEY `idx_audit_tenant_res`', '(`tenant_id`,`resource_type`,`resource_id`)'),
|
||
('KEY `idx_audit_trace`', '(`tenant_id`,`trace_id`)'),
|
||
],
|
||
},
|
||
'pbl_seed_record': {
|
||
'comment': 'PBL 幂等种子注入记录（appcodes/模板/治理种子防重）',
|
||
'columns': [
|
||
('tenant_id', "varchar(64) NOT NULL COMMENT '租户ID（* 表示全局种子）'"),
|
||
('seed_key', "varchar(128) NOT NULL COMMENT '种子键（模块:组:项）'"),
|
||
('seed_group', "varchar(64) DEFAULT NULL COMMENT '种子分组'"),
|
||
('module', "varchar(64) NOT NULL COMMENT '所属模块'"),
|
||
('payload', "longtext DEFAULT NULL COMMENT '种子内容 JSON'"),
|
||
('checksum', "varchar(64) DEFAULT NULL COMMENT '内容校验和（变更检测）'"),
|
||
('version', "int NOT NULL DEFAULT 1 COMMENT '注入版本'"),
|
||
('status', "varchar(16) NOT NULL DEFAULT 'applied' COMMENT 'applied/skipped/failed'"),
|
||
('created_at', "datetime NOT NULL COMMENT '创建时间'"),
|
||
('updated_at', "datetime NOT NULL COMMENT '更新时间'"),
|
||
],
|
||
'indexes': [
|
||
('PRIMARY KEY', '(`id`)'),
|
||
('UNIQUE KEY `uk_seed_tenant_key`', '(`tenant_id`,`seed_key`)'),
|
||
('KEY `idx_seed_module`', '(`tenant_id`,`module`)'),
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
def _assert_dialect(ddl_text, table=None):
|
||
"""方言自检：命中禁用 token 即抛错（mariadb 契约）"""
|
||
up = ddl_text.upper()
|
||
for token in FORBIDDEN_TOKENS:
|
||
if token.upper() in up:
|
||
raise DbError(
|
||
message='DDL 含禁用元素 %r（方言必须为 %s，表=%s）' % (token, DIALECT, table),
|
||
detail={'token': token, 'table': table, 'dialect': DIALECT},
|
||
)
|
||
return True
|
||
|
||
|
||
def ddl_of(table_name):
|
||
"""生成单表 CREATE TABLE 语句（mariadb 方言，IF NOT EXISTS 幂等）"""
|
||
spec = TABLES.get(table_name)
|
||
if not spec:
|
||
raise DbError(message='未登记的表：%s' % table_name, detail={'table': table_name})
|
||
|
||
cols = spec['columns']
|
||
if not cols or cols[0][0] != 'tenant_id':
|
||
raise DbError(
|
||
message='表 %s 首列必须为 tenant_id（实际 %s）' % (table_name, cols[0][0] if cols else None),
|
||
detail={'table': table_name},
|
||
)
|
||
|
||
lines = ["CREATE TABLE IF NOT EXISTS `%s` (" % table_name]
|
||
lines.append(" `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',")
|
||
for name, decl in cols:
|
||
lines.append(" `%s` %s," % (name, decl))
|
||
for idx_name, idx_decl in spec['indexes']:
|
||
lines.append(" %s %s," % (idx_name, idx_decl))
|
||
# 去掉最后一行逗号
|
||
lines[-1] = lines[-1].rstrip(',')
|
||
lines.append(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='%s';"
|
||
% spec.get('comment', table_name))
|
||
ddl = '\n'.join(lines)
|
||
_assert_dialect(ddl, table_name)
|
||
return ddl
|
||
|
||
|
||
def all_ddl():
|
||
"""生成全部公共表 DDL（拼接文本，供 build.sh / apply_ddl.sh 使用）"""
|
||
return '\n\n'.join(ddl_of(t) for t in sorted(TABLES.keys()))
|
||
|
||
|
||
def ensure_tables(conn=None, module='pbl_common'):
|
||
"""
|
||
幂等建表（CREATE TABLE IF NOT EXISTS）。
|
||
返回已确保的表名列表。conn 为 None 时自建连接。
|
||
"""
|
||
from pbl_common.dbutil import execute, get_conn
|
||
|
||
own = conn is None
|
||
conn = conn or get_conn(module=module)
|
||
created = []
|
||
try:
|
||
for table in sorted(TABLES.keys()):
|
||
execute(ddl_of(table), conn=conn)
|
||
created.append(table)
|
||
if own:
|
||
try:
|
||
conn.commit()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
finally:
|
||
if own:
|
||
try:
|
||
conn.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return created
|
||
|
||
|
||
def table_names():
|
||
"""公共表名清单"""
|
||
return sorted(TABLES.keys())
|