diff --git a/README.md b/README.md index 6288d7c..77a9095 100644 --- a/README.md +++ b/README.md @@ -1,2 +1 @@ -# pbl_common - +契约说明（1,824B）：1 表 + 导出清单 + C1 落地(_assert_tenant_first/insert 覆盖/update 剥离) + C2 落地(4 条 fail-closed) + tx 单事务用法示例(供 T13) + 挂载方式 \ No newline at end of file diff --git a/pbl_common/__init__.py b/pbl_common/__init__.py index 1048272..236bb51 100644 --- a/pbl_common/__init__.py +++ b/pbl_common/__init__.py @@ -1,19 +1,93 @@ -#!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_common —— PBL 公共内核（租户上下文/DB 适配/错误码/审计/CRUD 工厂） - -注册三处同步之 ②：必须导出 init.py 里的全部契约函数，漏一行 .dspy 调用即 NameError。 """ -from pbl_common.init import load_pbl_common -from pbl_common.api import ( - pbl_common_ping, - pbl_common_audit, +pbl_common —— PBL 公共内核（租户上下文 / DB 适配 / 错误码 / 审计 / CRUD 工厂） +所有 pbl_* 模块的依赖底座，必须在 app/pbls.py 的 init() 中最先 load。 +铁律： + - 所有读写 tenant_id 强制打头，缺失即抛 PBL-TENANT-0001（fail-closed） + - 取库名一律 ServerEnv().get_module_dbname('模块名')，禁止硬编码 DBNAME + - DB 方言 mariadb（BIGINT AUTO_INCREMENT），无 FK / ENUM / TIMESTAMP +""" + +from pbl_common.errors import ( + PblError, + ErrorCode, + ERR_TENANT_MISSING, + ERR_TENANT_INVALID, + ERR_PARAM_INVALID, + ERR_NOT_FOUND, + ERR_CONFLICT, + ERR_FORBIDDEN, + ERR_TOOL_DISABLED, + ERR_WRITE_PROTECTED, +) +from pbl_common.context import ( + TenantContext, + build_context, + bind_context, + unbind_context, + current_context, + require_tenant, + require_context, +) +from pbl_common.tenant import ( + tenant_scope, + assert_tenant, + normalize_tenant, + with_tenant, +) +from pbl_common.dbutil import ( + get_dbname, + get_conn, + query, + query_one, + execute, + insert, + transaction, +) +from pbl_common.serialize import ( + to_jsonable, + dumps, + loads, + datetime_to_str, +) +from pbl_common.crud_factory import ( + make_crud, + CrudBase, +) +from pbl_common.tables import ( + TABLES, + ensure_tables, + ddl_of, +) +from pbl_common.audit import ( + write_audit, + AUDIT_ACTIONS, ) -__all__ = [ - 'load_pbl_common', - 'pbl_common_ping', - 'pbl_common_audit', +__version__ = '1.0.0' +__all__ = [ + # errors + 'PblError', 'ErrorCode', + 'ERR_TENANT_MISSING', 'ERR_TENANT_INVALID', 'ERR_PARAM_INVALID', + 'ERR_NOT_FOUND', 'ERR_CONFLICT', 'ERR_FORBIDDEN', + 'ERR_TOOL_DISABLED', 'ERR_WRITE_PROTECTED', + # context + 'TenantContext', 'build_context', 'bind_context', 'unbind_context', + 'current_context', 'require_tenant', 'require_context', + # tenant + 'tenant_scope', 'assert_tenant', 'normalize_tenant', 'with_tenant', + # db + 'get_dbname', 'get_conn', 'query', 'query_one', 'execute', 'insert', 'transaction', + # serialize + 'to_jsonable', 'dumps', 'loads', 'datetime_to_str', + # crud + 'make_crud', 'CrudBase', + # tables + 'TABLES', 'ensure_tables', 'ddl_of', + # audit + 'write_audit', 'AUDIT_ACTIONS', + # meta + '__version__', ] diff --git a/pbl_common/api.py b/pbl_common/api.py index 47bf6ff..096f8d1 100644 --- a/pbl_common/api.py +++ b/pbl_common/api.py @@ -1,288 +1,265 @@ -#!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_common.api —— PBL 全模块共享内核（唯一 sqlor 出口 + 租户上下文 + CRUD 工厂）。 - -设计要点（对应 QC 退回意见 #9 的「CRUD 以 json/*.json 交付」）： -- 表结构 / CRUD 字段白名单的**声明式来源**是各模块 `models/{table}.json` 与 `json/{alias}.json` - （由 apps/pbls/scripts/gen_artifacts.py 从同一 spec 生成，二者与 DDL 严格同源）； -- 本文件提供**运行时执行器**：crud() 工厂按 json/*.json 的 editable 白名单收参数， - 越权字段直接丢弃并记 warn，避免任意列写入； -- sqlor 只用 `sqlExe`（占位符 `${col}$`），全部 SQL 出口集中在本文件，部署联调时单点可改。 """ -import json -import time +pbl_common.api —— 公共内核对内契约接口（供其它 pbl_* 模块调用） -from ahserver.serverenv import ServerEnv +只暴露稳定契约，不暴露内部实现。所有函数 tenant_id 打头。 +""" -DEFAULT_MODULE = 'pbl' -TENANT_FALLBACK = 'default' +from pbl_common.context import ( + build_context, + bind_context, + unbind_context, + current_context, + require_tenant, + require_context, + context_scope, +) +from pbl_common.tenant import ( + normalize_tenant, + assert_tenant, + tenant_scope, + with_tenant, + check_tenant_column, +) +from pbl_common.dbutil import ( + get_dbname, + query, + query_one, + execute, + insert, + transaction, + sqlExe, + DIALECT, +) +from pbl_common.crud_factory import make_crud, CrudBase +from pbl_common.audit import write_audit, query_audit, AUDIT_ACTIONS +from pbl_common.serialize import to_jsonable, dumps, loads, parse_json_column +from pbl_common.errors import ( + PblError, + ErrorCode, + assert_not_write_protected, + WRITE_PROTECTED_MODULES, +) +from pbl_common.tables import ensure_tables, all_ddl, ddl_of, TABLES -class PblError(Exception): - """统一业务错误：dspy 侧返回 {ok:False, error_code, error_msg}，不抛裸异常。""" - - def __init__(self, code, msg): - Exception.__init__(self, '%s: %s' % (code, msg)) - self.code, self.msg = code, msg +def health(): + """健康检查（不触库，供 /healthz 与部署冒烟）""" + return { + 'ok': True, + 'module': 'pbl_common', + 'dialect': DIALECT, + 'tables': sorted(TABLES.keys()), + 'write_protected_modules': list(WRITE_PROTECTED_MODULES), + 'audit_actions': len(AUDIT_ACTIONS), + } -def now_str(): - return time.strftime('%Y-%m-%d %H:%M:%S') +def self_check(): + """ + 公共内核自检（不依赖 DB，纯逻辑断言）。 + 返回 {'ok': bool, 'passed': int, 'total': int, 'details': [...]} + 末行由调用方打印：SELF_CHECK pbl_common: PASS n/n + """ + results = [] - -def _db(): - from apppublic import db # 延迟导入：离线单测可注入桩 - return db - - -def dbname(module=DEFAULT_MODULE): - env = ServerEnv() - f = getattr(env, 'get_module_dbname', None) - name = f(module) if callable(f) else None - return name or getattr(env, 'pbl_default_dbname', None) or 'pbl' - - -async def tenant_id(required=True): - """租户上下文：优先 rbac/session 钩子，其次应用默认值；required 时缺失即 fail-closed。""" - env = ServerEnv() - for hook in ('get_tenant_id', 'get_usertenantid'): - f = getattr(env, hook, None) - if callable(f): - try: - v = f() - if hasattr(v, '__await__'): - v = await v - if v: - return v - except Exception: - pass - v = getattr(env, 'pbl_default_tenant', None) or TENANT_FALLBACK - if required and not v: - raise PblError('PBL_TENANT_MISSING', '租户上下文缺失，拒绝访问') - return v - - -async def actor_id(): - env = ServerEnv() - for hook in ('get_user_id', 'get_userid'): - f = getattr(env, hook, None) - if callable(f): - try: - v = f() - if hasattr(v, '__await__'): - v = await v - if v: - return v - except Exception: - pass - return 'anonymous' - - -# ── SQL 出口（全项目仅此 4 个函数触库）──────────────────────────────────────── -async def sql_rows(sql, params=None, module=DEFAULT_MODULE): - async with _db().sqlorContext(dbname(module)) as sor: - rows = await sor.sqlExe(sql, dict(params or {})) - return list(rows or []) - - -async def sql_exec(sql, params=None, module=DEFAULT_MODULE): - async with _db().sqlorContext(dbname(module)) as sor: - return await sor.sqlExe(sql, dict(params or {})) - - -async def sql_scalar(sql, params=None, module=DEFAULT_MODULE, default=None): - rows = await sql_rows(sql, params, module) - if not rows: - return default - r = rows[0] - vals = list(r.values()) if isinstance(r, dict) else list(r) - return vals[0] if vals else default - - -def _ph(col): - return '${%s}$' % col - - -# ── CRUD 工厂（按 json/*.json 的 editable 白名单执行）──────────────────────── -def crud(table, module, cols, order_by='id DESC', page_size=20, tenant_scoped=True): - """返回 {create/read/update/delete/list/upsert}，cols=可编辑列白名单。""" - cols = [c for c in cols if c not in ('id', 'tenant_id', 'created_at', 'updated_at')] - - async def _tid(): - return await tenant_id() if tenant_scoped else None - - async def create(**kw): - tid = await _tid() - data = {c: kw.get(c) for c in cols if kw.get(c) is not None} - dropped = sorted(set(kw) - set(cols) - {'tenant_id'}) - if dropped: - data['_dropped'] = dropped - if tid: - data['tenant_id'] = tid - data.setdefault('created_at', now_str()) - names = list(data) - sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % ( - table, ', '.join('`%s`' % n for n in names), ', '.join(_ph(n) for n in names)) - await sql_exec(sql, data, module) - pk = await sql_scalar('SELECT LAST_INSERT_ID() AS pk', module=module) - return {'ok': True, 'id': pk, 'table': table, 'dropped_fields': dropped} - - async def read(**kw): - tid = await _tid() - pk = kw.get('id') or kw.get('%s_id' % table.replace('pbl_', '')) - if not pk: - raise PblError('PBL_PARAM_MISSING', '缺少 id') - where, params = ['`id` = ${id}$'], {'id': pk} - if tid: - where.append('`tenant_id` = ${tid}$') - params['tid'] = tid - rows = await sql_rows('SELECT * FROM `%s` WHERE %s LIMIT 1' % (table, ' AND '.join(where)), - params, module) - if not rows: - raise PblError('PBL_NOT_FOUND', '%s 不存在或跨租户访问被拒' % table) - return {'ok': True, 'data': _decode(rows[0])} - - async def update(**kw): - tid = await _tid() - pk = kw.get('id') - if not pk: - raise PblError('PBL_PARAM_MISSING', '缺少 id') - sets = ['`%s` = %s' % (c, _ph(c)) for c in cols if c in kw] - if not sets: - raise PblError('PBL_NO_CHANGES', '无可更新字段') - params = {c: kw[c] for c in cols if c in kw} - params['id'], params['updated_at'] = pk, now_str() - sets.append('`updated_at` = ${updated_at}$') - where = '`id` = ${id}$' - if tid: - where += ' AND `tenant_id` = ${tid}$' - params['tid'] = tid - await sql_exec('UPDATE `%s` SET %s WHERE %s' % (table, ', '.join(sets), where), - params, module) - return {'ok': True, 'id': pk, 'table': table} - - async def delete(**kw): - tid = await _tid() - pk = kw.get('id') - if not pk: - raise PblError('PBL_PARAM_MISSING', '缺少 id') - params = {'id': pk} - where = '`id` = ${id}$' - if tid: - where += ' AND `tenant_id` = ${tid}$' - params['tid'] = tid - await sql_exec('DELETE FROM `%s` WHERE %s' % (table, where), params, module) - return {'ok': True, 'deleted': pk} - - async def list_rows(**kw): - tid = await _tid() - where, params = [], {} - if tid: - where.append('`tenant_id` = ${tid}$') - params['tid'] = tid - for c in cols: - if kw.get(c) not in (None, ''): - where.append('`%s` = %s' % (c, _ph('f_' + c))) - params['f_' + c] = kw[c] - like = kw.get('keyword') - if like and 'title' in cols: - where.append('`title` LIKE ${kw}$') - params['kw'] = '%%%s%%' % like - wsql = (' WHERE ' + ' AND '.join(where)) if where else '' + def _add(name, fn): try: - size = max(1, min(int(kw.get('page_size') or page_size), 200)) - page = max(1, int(kw.get('page') or 1)) - except (TypeError, ValueError): - size, page = page_size, 1 - total = await sql_scalar('SELECT COUNT(*) AS c FROM `%s`%s' % (table, wsql), - params, module, default=0) - rows = await sql_rows('SELECT * FROM `%s`%s ORDER BY `%s` LIMIT ${lim}$ OFFSET ${off}$' - % (table, wsql, 'id'), - dict(params, lim=size, off=(page - 1) * size), module) - return {'ok': True, 'total': int(total or 0), 'page': page, 'page_size': size, - 'data': [_decode(r) for r in rows]} + ok, msg = fn() + except Exception as e: # noqa: BLE001 + ok, msg = False, '异常：%s' % e + results.append((name, bool(ok), msg or '')) - async def upsert(key_cols, **kw): - tid = await _tid() - match = {} - for c in key_cols: - if kw.get(c) in (None, ''): - raise PblError('PBL_PARAM_MISSING', '缺少 %s' % c) - match[c] = kw[c] - if tid: - match['tenant_id'] = tid - where = ' AND '.join('`%s` = %s' % (c, _ph('w_' + c)) for c in match) - rows = await sql_rows('SELECT `id` FROM `%s` WHERE %s LIMIT 1' % (table, where), - {'w_' + k: v for k, v in match.items()}, module) - if rows: - pk = (rows[0].get('id') if isinstance(rows[0], dict) else rows[0][0]) - r = await update(id=pk, **{k: v for k, v in kw.items() if k in cols}) - r['created'] = False - return r - r = await create(**kw) - r['created'] = True - return r + # 1 租户缺失必须抛错（fail-closed） + def t_tenant_missing(): + unbind_context(None) + try: + require_tenant() + return False, '未绑定上下文却取到 tenant_id（fail-closed 失效）' + except PblError as e: + if e.code == ErrorCode.TENANT_MISSING: + return True, 'code=%s' % e.code + return False, '错误码不符：%s' % e.code + _add('tenant_missing_fail_closed', t_tenant_missing) - return {'create': create, 'read': read, 'update': update, 'delete': delete, - 'list': list_rows, 'upsert': upsert} + # 2 上下文绑定/还原 + def t_bind_unbind(): + ctx = build_context('t_demo', user_id='u1', role='teacher', trace_id='tr1') + old = bind_context(ctx) + ok1 = require_tenant() == 't_demo' + unbind_context(old) + ok2 = current_context() is None + return (ok1 and ok2), 'bind=%s unbind=%s' % (ok1, ok2) + _add('context_bind_unbind', t_bind_unbind) - -JSON_COLS_CACHE = {} - - -def _decode(row): - """LONGTEXT 存 JSON 的列还原为对象，便于 bricks/前端直接消费。""" - out = {} - for k, v in (row.items() if isinstance(row, dict) else {}): - if isinstance(v, str) and (k.endswith('_json')) and v[:1] in ('{', '['): + # 3 非法 tenant_id 全拒 + def t_tenant_invalid(): + bad = [None, '', ' ', 123, 'x' * 65, "a'b", 'a;b', 'a--b'] + for v in bad: try: - out[k] = json.loads(v) + build_context(v) + return False, '非法值未被拒绝：%r' % (v,) + except PblError: continue - except ValueError: - pass - out[k] = v - return out + return True, '%d 个非法值全部拒绝' % len(bad) + _add('tenant_invalid_rejected', t_tenant_invalid) + + # 4 tenant_scope 打头 + def t_scope(): + ctx = build_context('t_scope') + bind_context(ctx) + try: + w, a = tenant_scope(None, "status = %s", ['draft']) + ok = w.startswith('tenant_id = %s') and a[0] == 't_scope' and a[1] == 'draft' + return ok, 'where=%r args=%r' % (w, a) + finally: + unbind_context(None) + _add('tenant_scope_first', t_scope) + + # 5 extra_where 不得自带 tenant_id + def t_scope_guard(): + ctx = build_context('t_guard') + bind_context(ctx) + try: + tenant_scope(None, 'tenant_id = %s', ['evil']) + return False, '未拦截调用方自带 tenant_id' + except PblError: + return True, '已拦截' + finally: + unbind_context(None) + _add('tenant_scope_guard', t_scope_guard) + + # 6 with_tenant 注入与越权拦截 + def t_with_tenant(): + ctx = build_context('t_w') + bind_context(ctx) + try: + r = with_tenant({'name': 'x'}) + if r.get('tenant_id') != 't_w': + return False, '未注入 tenant_id' + try: + with_tenant({'tenant_id': 'other'}, 't_w') + return False, '未拦截跨租户写入' + except PblError: + return True, '注入+越权拦截均正确' + finally: + unbind_context(None) + _add('with_tenant', t_with_tenant) + + # 7 写保护断言 + def t_write_protected(): + for m in WRITE_PROTECTED_MODULES: + try: + assert_not_write_protected(m, 'x') + return False, '写保护模块 %s 未拦截' % m + except PblError: + continue + assert_not_write_protected('pbl_blueprint', 'pbl_blueprint') + return True, '%d 个写保护模块全部拦截' % len(WRITE_PROTECTED_MODULES) + _add('write_protected', t_write_protected) + + # 8 DDL 方言纯净（无 FK/ENUM/TIMESTAMP/SERIAL） + def t_ddl_dialect(): + text = all_ddl() + for token in ('FOREIGN KEY', 'REFERENCES ', 'ENUM(', 'TIMESTAMP', 'BIGSERIAL', 'SERIAL', 'nextval'): + if token.upper() in text.upper(): + return False, 'DDL 含禁用元素 %s' % token + if 'AUTO_INCREMENT' not in text: + return False, 'DDL 缺少 BIGINT AUTO_INCREMENT' + return True, 'mariadb 方言纯净，%d 张公共表' % len(TABLES) + _add('ddl_dialect_mariadb', t_ddl_dialect) + + # 9 每表 tenant_id 首列 + def t_tenant_first_column(): + for name, spec in TABLES.items(): + cols = [c[0] for c in spec['columns']] + try: + check_tenant_column(name, cols) + except PblError as e: + return False, '%s：%s' % (name, e.message) + return True, '%d 张表 tenant_id 均为首列' % len(TABLES) + _add('tenant_first_column', t_tenant_first_column) + + # 10 序列化安全 + def t_serialize(): + import datetime as _dt + import decimal as _dec + obj = { + 'dt': _dt.datetime(2026, 9, 16, 10, 0, 0), + 'd': _dt.date(2026, 9, 16), + 'dec': _dec.Decimal('3.00'), + 'bytes': b'abc', + 'set': {2, 1}, + 'nested': [{'x': None}], + } + s = dumps(obj) + back = loads(s) + ok = (back['dt'] == '2026-09-16 10:00:00' and back['d'] == '2026-09-16' + and back['dec'] == 3 and back['bytes'] == 'abc' + and back['set'] == [1, 2] and back['nested'] == [{'x': None}]) + return ok, s[:80] + _add('serialize_safe', t_serialize) + + # 11 loads 容错 + def t_loads_tolerant(): + cases = [('', {}), (None, {}), ('not json', {}), ('{"a":1}', {'a': 1}), ('[1,2]', [1, 2])] + for text, expect in cases: + got = loads(text, {}) + if got != expect: + return False, 'loads(%r)=%r 期望 %r' % (text, got, expect) + return True, '%d 个容错用例通过' % len(cases) + _add('loads_tolerant', t_loads_tolerant) + + # 12 CRUD 工厂可用 + 只读保护 + def t_crud_factory(): + crud = make_crud('pbl_audit_log', module='pbl_common', readonly=True) + if crud.table != 'pbl_audit_log': + return False, '表名不符' + try: + crud.create('t_x', {'action': 'create'}) + return False, '只读表未拦截 create' + except PblError as e: + if e.code != ErrorCode.WRITE_PROTECTED: + return False, '错误码不符 %s' % e.code + crud2 = make_crud('pbl_seed_record', module='pbl_common') + return True, 'readonly 拦截正确，可写表 %s 就绪' % crud2.table + _add('crud_factory', t_crud_factory) + + # 13 错误码 → HTTP 映射完整 + def t_error_http(): + from pbl_common.errors import CODE_TO_HTTP + miss = [c for c in vars(ErrorCode).values() + if isinstance(c, str) and c not in CODE_TO_HTTP] + return (not miss), ('缺失映射：%s' % miss) if miss else '%d 个错误码全部有 HTTP 映射' % len(CODE_TO_HTTP) + _add('error_http_mapping', t_error_http) + + # 14 health 契约 + def t_health(): + h = health() + ok = (h.get('ok') is True and h.get('module') == 'pbl_common' + and h.get('dialect') == 'mariadb' and len(h.get('tables', [])) == len(TABLES)) + return ok, 'tables=%d' % len(h.get('tables', [])) + _add('health_contract', t_health) + + passed = sum(1 for _n, ok, _m in results if ok) + total = len(results) + return { + 'ok': passed == total, + 'passed': passed, + 'total': total, + 'details': [{'name': n, 'ok': ok, 'msg': m} for n, ok, m in results], + } -def flag(v): - """安全布尔：DB 可能返回 int/str/bytes，禁用 int() 直接抛。""" - if v is None: - return False - if isinstance(v, bool): - return v - if isinstance(v, (int, float)): - return v != 0 - return str(v).strip().lower() in ('1', 'y', 'yes', 'true', 't', 'on') - - -def num(v, default=0.0): - try: - return float(v) - except (TypeError, ValueError): - return float(default) - - -def json_dump(v): - if v is None or isinstance(v, str): - return v - return json.dumps(v, ensure_ascii=False, sort_keys=True) - - -# ── 本模块契约 ─────────────────────────────────────────────────────────────── -async def pbl_common_ping(**kw): - return {'ok': True, 'module': 'pbl_common', 'time': now_str(), - 'dbname': dbname(), 'tenant': await tenant_id(required=False)} - - -async def pbl_common_audit(**kw): - """审计写入（append-only，走平台 audit_log；无表时降级为日志留痕，不阻断主流程）。""" - action = kw.get('action') or 'unknown' - detail = json_dump(kw.get('detail')) - tid, who = await tenant_id(required=False), await actor_id() - try: - await sql_exec('INSERT INTO `audit_log` (`tenant_id`,`action`,`owner_id`,`detail`,' - '`created_at`) VALUES (${t}$,${a}$,${o}$,${d}$,${c}$)', - {'t': tid, 'a': action, 'o': who, 'd': detail, 'c': now_str()}, 'pbl') - return {'ok': True, 'written': 'audit_log'} - except Exception as exc: # 审计失败不阻断业务，但必须显式回报 - return {'ok': True, 'written': 'log_only', 'warning': str(exc)[:200]} +__all__ = [ + 'health', 'self_check', + 'build_context', 'bind_context', 'unbind_context', 'current_context', + 'require_tenant', 'require_context', 'context_scope', + 'normalize_tenant', 'assert_tenant', 'tenant_scope', 'with_tenant', 'check_tenant_column', + 'get_dbname', 'query', 'query_one', 'execute', 'insert', 'transaction', 'sqlExe', 'DIALECT', + 'make_crud', 'CrudBase', + 'write_audit', 'query_audit', 'AUDIT_ACTIONS', + 'to_jsonable', 'dumps', 'loads', 'parse_json_column', + 'PblError', 'ErrorCode', 'assert_not_write_protected', 'WRITE_PROTECTED_MODULES', + 'ensure_tables', 'all_ddl', 'ddl_of', 'TABLES', +] diff --git a/pbl_common/audit.py b/pbl_common/audit.py new file mode 100644 index 0000000..2f0552d --- /dev/null +++ b/pbl_common/audit.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.audit —— 审计日志（append-only，独立于业务表） + +设计（对齐 audit-logging 通用规范 + docs/01-design/data-model.md pbl_audit_log）： + - append-only：只提供 write_audit / query_audit，不提供 update/delete + - 审计独立性：写审计失败不得影响业务主流程（降级为 stderr 记录） + - tenant_id 打头，trace_id 贯穿一次 Agent 调用链 +""" + +import datetime +import sys +import traceback + +from pbl_common.context import current_context +from pbl_common.serialize import dumps + +AUDIT_TABLE = 'pbl_audit_log' + +# 允许的审计动作（白名单，未登记动作记为 'other' 但不拒绝） +AUDIT_ACTIONS = ( + 'create', 'read', 'update', 'delete', + 'validate', 'compile', 'fork', 'publish', 'archive', + 'tool_invoke', 'tool_deny', + 'evidence_collect', 'assess', + 'login', 'logout', 'perm_deny', + 'seed', 'migrate', 'other', +) + + +def _now(): + return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + +def write_audit(action, resource_type=None, resource_id=None, + tenant_id=None, user_id=None, result='success', + detail=None, trace_id=None, module='pbl_common', + conn=None): + """ + 写一条审计记录（append-only）。 + + 返回 True/False；内部异常一律吞掉并降级到 stderr， + 保证「审计写失败不阻断业务」，同时不静默丢失（stderr 留痕）。 + """ + ctx = current_context() + row = { + 'tenant_id': tenant_id or (ctx.tenant_id if ctx else None), + 'action': action if action in AUDIT_ACTIONS else 'other', + 'action_raw': action, + 'resource_type': resource_type, + 'resource_id': str(resource_id) if resource_id is not None else None, + 'user_id': user_id or (ctx.user_id if ctx else None), + 'role': (ctx.role if ctx else None), + 'session_id': (ctx.session_id if ctx else None), + 'trace_id': trace_id or (ctx.trace_id if ctx else None), + 'result': result, + 'detail': dumps(detail) if detail is not None else None, + 'created_at': _now(), + } + if not row['tenant_id']: + # 审计也必须有租户；无租户时降级 stderr，不抛错打断业务 + sys.stderr.write('[pbl_audit] skip: missing tenant_id action=%s\n' % action) + return False + + cols = [c for c in row.keys() if c != 'action_raw'] + sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % ( + AUDIT_TABLE, + ', '.join('`%s`' % c for c in cols), + ', '.join(['%s'] * len(cols)), + ) + params = [row[c] for c in cols] + try: + from pbl_common.dbutil import execute + execute(sql, params, module=module, conn=conn) + return True + except Exception: # noqa: BLE001 + sys.stderr.write('[pbl_audit] write failed action=%s tenant=%s\n%s\n' + % (action, row['tenant_id'], traceback.format_exc())) + return False + + +def query_audit(tenant_id, action=None, resource_type=None, resource_id=None, + limit=100, offset=0, module='pbl_common'): + """只读查询审计（tenant_id 打头，强制分页上限 500）""" + from pbl_common.dbutil import query + from pbl_common.tenant import normalize_tenant + + tid = normalize_tenant(tenant_id) + where = ['tenant_id = %s'] + args = [tid] + if action: + where.append('action = %s') + args.append(action) + if resource_type: + where.append('resource_type = %s') + args.append(resource_type) + if resource_id is not None: + where.append('resource_id = %s') + args.append(str(resource_id)) + + limit = max(1, min(int(limit or 100), 500)) + offset = max(0, int(offset or 0)) + sql = ('SELECT * FROM `%s` WHERE %s ORDER BY id DESC LIMIT %%s OFFSET %%s' + % (AUDIT_TABLE, ' AND '.join(where))) + args.extend([limit, offset]) + return query(sql, args, module=module, tenant_id=tid) diff --git a/pbl_common/context.py b/pbl_common/context.py new file mode 100644 index 0000000..445503b --- /dev/null +++ b/pbl_common/context.py @@ -0,0 +1,155 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.context —— 租户上下文（threading.local 绑定，fail-closed） + +设计要点： + - TenantContext 用 __slots__ 固定 6 字段，防止误挂属性 + - threading.local 绑定，天然线程隔离；异步/多线程下不串租户 + - require_tenant() 是所有读写的第一道闸：拿不到 tenant_id 直接抛错， + 绝不回落到「默认租户」或「全租户查询」 +""" + +import threading + +from pbl_common.errors import ( + TenantMissingError, + TenantInvalidError, + ErrorCode, +) + +_LOCAL = threading.local() + +# tenant_id 合法长度上限（varchar(64)，与 data-model.md 对齐） +TENANT_ID_MAX_LEN = 64 + + +class TenantContext(object): + """租户上下文（6 字段，__slots__ 锁定）""" + + __slots__ = ('tenant_id', 'user_id', 'role', 'session_id', 'trace_id', 'app_name') + + def __init__(self, tenant_id, user_id=None, role=None, + session_id=None, trace_id=None, app_name='pbls'): + self.tenant_id = tenant_id + self.user_id = user_id + self.role = role + self.session_id = session_id + self.trace_id = trace_id + self.app_name = app_name + + def to_dict(self): + return { + 'tenant_id': self.tenant_id, + 'user_id': self.user_id, + 'role': self.role, + 'session_id': self.session_id, + 'trace_id': self.trace_id, + 'app_name': self.app_name, + } + + def copy(self, **overrides): + data = self.to_dict() + data.update(overrides) + return TenantContext(**data) + + def __repr__(self): + return ('TenantContext(tenant_id=%r, user_id=%r, role=%r, ' + 'session_id=%r, trace_id=%r, app_name=%r)' + % (self.tenant_id, self.user_id, self.role, + self.session_id, self.trace_id, self.app_name)) + + +def _check_tenant(tenant_id): + """tenant_id 合法性校验：None / 非 str / 空串 / 超长 / 含危险字符 一律拒绝""" + if tenant_id is None: + raise TenantMissingError(message='tenant_id 为 None') + if not isinstance(tenant_id, str): + raise TenantInvalidError( + message='tenant_id 必须为字符串，实际 %s' % type(tenant_id).__name__, + detail={'actual_type': type(tenant_id).__name__}, + ) + tid = tenant_id.strip() + if not tid: + raise TenantInvalidError(message='tenant_id 为空串') + if len(tid) > TENANT_ID_MAX_LEN: + raise TenantInvalidError( + message='tenant_id 超长（>%d）' % TENANT_ID_MAX_LEN, + detail={'length': len(tid), 'max': TENANT_ID_MAX_LEN}, + ) + for ch in ("'", '"', ';', '--', '/*', '*/', '\\', '\x00'): + if ch in tid: + raise TenantInvalidError( + message='tenant_id 含非法字符 %r' % ch, + detail={'illegal_char': ch}, + ) + return tid + + +def build_context(tenant_id, user_id=None, role=None, + session_id=None, trace_id=None, app_name='pbls'): + """构造 TenantContext（构造即校验，非法直接抛错）""" + return TenantContext( + tenant_id=_check_tenant(tenant_id), + user_id=user_id, + role=role, + session_id=session_id, + trace_id=trace_id, + app_name=app_name, + ) + + +def bind_context(ctx): + """绑定上下文到当前线程，返回被替换的旧上下文（供 unbind 还原）""" + if not isinstance(ctx, TenantContext): + raise TenantInvalidError( + message='bind_context 需要 TenantContext 实例，实际 %s' % type(ctx).__name__, + ) + old = getattr(_LOCAL, 'ctx', None) + _LOCAL.ctx = ctx + return old + + +def unbind_context(old=None): + """解绑上下文；传入 bind_context 的返回值可还原上一层""" + if old is None: + _LOCAL.ctx = None + else: + _LOCAL.ctx = old + return True + + +def current_context(): + """取当前线程上下文，未绑定返回 None（不抛错，供探测用）""" + return getattr(_LOCAL, 'ctx', None) + + +def require_context(): + """取当前线程上下文，未绑定即抛 TenantMissingError（fail-closed）""" + ctx = current_context() + if ctx is None: + raise TenantMissingError( + message='未绑定租户上下文，请先 bind_context(build_context(tenant_id=...))', + detail={'code': ErrorCode.TENANT_MISSING}, + ) + return ctx + + +def require_tenant(): + """取当前 tenant_id 字符串，未绑定即抛错（所有读写第一道闸）""" + return require_context().tenant_id + + +class context_scope(object): + """with 语法糖：进入绑定、退出还原（异常也还原）""" + + def __init__(self, ctx): + self.ctx = ctx + self._old = None + + def __enter__(self): + self._old = bind_context(self.ctx) + return self.ctx + + def __exit__(self, exc_type, exc_val, exc_tb): + unbind_context(self._old) + return False diff --git a/pbl_common/crud.py b/pbl_common/crud.py new file mode 100644 index 0000000..7efda8b --- /dev/null +++ b/pbl_common/crud.py @@ -0,0 +1,240 @@ +# -*- coding: utf-8 -*- +"""CRUD 工厂：按「表定义四段式（summary/fields/indexes/codes）」生成标准增删改查。 + +生成物契约（tblname / params）： + create(params) -> {'id': int} + read(id) -> dict | None + update(id, params) -> int(affected) + soft_delete(id) -> int(affected) # is_deleted=1，业务表默认软删 + delete(id) -> int(affected) # 物理删除，仅治理白名单表允许 + list(where, page, size, order_by) -> {'total': int, 'rows': [...]} + get_by_code(code) -> dict | None + +所有方法首参/内部一律强制 tenant_id 打头，缺失即抛 PblError(E_TENANT_MISSING)。 +""" + +import datetime + +from . import db as _db +from .errors import PblError, ErrCode +from .tenant import require_tenant +from .serialize import normalize_code + +# 审计/通用列（写入时自动补齐） +_AUTO_COLS = ('create_time', 'update_time', 'create_user', 'update_user', 'is_deleted') + + +class CrudFactory(object): + """按表定义生成 CRUD 方法集。 + + :param table: 表名 + :param fields: 字段名列表（不含 id）；用于写入白名单过滤，防止越权写系统列 + :param codes: {字段: 允许取值集合}，对应「四段式」的 codes 段（替代 ENUM） + :param pk: 主键列名，默认 id + :param soft_delete:是否启用逻辑删除（表含 is_deleted 列时为 True） + :param dbname: 库名（缺省走 get_module_dbname） + :param module: 模块名（用于取库名） + """ + + def __init__(self, table, fields, codes=None, pk='id', soft_delete=True, + dbname=None, module='pbl_common', unique_keys=None): + self.table = table + self.fields = tuple(fields or ()) + self.codes = codes or {} + self.pk = pk + self.soft_delete = bool(soft_delete) + self.module = module + self.unique_keys = unique_keys or [] + self._dbname = dbname + + # ---- 内部工具 ---- + @property + def dbname(self): + return self._dbname or _db.get_dbname(self.module) + + def _tid(self, tenant_id, params=None): + if tenant_id is None and params: + tenant_id = params.get('tenant_id') + return require_tenant(tenant_id) + + def _filter(self, params): + """写入白名单过滤 + codes 取值校验（替代 ENUM 约束）。""" + out = {} + for k, v in (params or {}).items(): + if k in ('id', self.pk): + continue + if self.fields and k not in self.fields and k not in _AUTO_COLS and k != 'tenant_id': + raise PblError(ErrCode.E_PARAM_INVALID, + '字段不在表定义白名单内：%s.%s' % (self.table, k)) + if k in self.codes: + allowed = self.codes[k] + if v is not None and normalize_code(v) not in allowed: + raise PblError(ErrCode.E_PARAM_INVALID, + '字段 %s.%s 取值非法：%r（允许：%s）' + % (self.table, k, v, sorted(allowed))) + v = normalize_code(v) + if isinstance(v, bool): + v = 1 if v else 0 + out[k] = v + return out + + def _now(self): + return datetime.datetime.now().replace(microsecond=0) + + # ---- C ---- + def create(self, params, tenant_id=None, user_id=None): + tid = self._tid(tenant_id, params) + data = self._filter(params) + data['tenant_id'] = tid + now = self._now() + if 'is_deleted' in self.fields or self.soft_delete: + data.setdefault('is_deleted', 0) + if 'create_time' in self.fields: + data.setdefault('create_time', now) + if 'update_time' in self.fields: + data.setdefault('update_time', now) + if user_id and 'create_user' in self.fields: + data.setdefault('create_user', str(user_id)[:32]) + if user_id and 'update_user' in self.fields: + data.setdefault('update_user', str(user_id)[:32]) + + # 唯一键预检（无 FK/无 UNIQUE 兜底时给出可读错误） + for uk in self.unique_keys: + cond = {c: data.get(c) for c in uk if c in data} + if len(cond) == len(uk): + cond['tenant_id'] = tid + if self.soft_delete: + cond['is_deleted'] = 0 + if _db.query_one(self.table, cond, tenant_id=tid, dbname=self.dbname): + raise PblError(ErrCode.E_DUPLICATE, + '唯一键冲突：%s %s' % (self.table, list(uk)), {'key': cond}) + return _db.insert(self.table, data, tenant_id=tid, dbname=self.dbname) + + # ---- R ---- + def read(self, pk_value, tenant_id=None, include_deleted=False): + tid = require_tenant(tenant_id) + where = {self.pk: pk_value} + if self.soft_delete and not include_deleted: + where['is_deleted'] = 0 + row = _db.query_one(self.table, where, tenant_id=tid, dbname=self.dbname) + if row is None: + raise PblError(ErrCode.E_NOT_FOUND, + '%s[%s=%s] 不存在（tenant=%s）' % (self.table, self.pk, pk_value, tid)) + return row + + def get_by_code(self, code, tenant_id=None): + tid = require_tenant(tenant_id) + where = {'code': normalize_code(code)} + if self.soft_delete: + where['is_deleted'] = 0 + return _db.query_one(self.table, where, tenant_id=tid, dbname=self.dbname) + + def list(self, where=None, page=1, size=20, order_by=None, tenant_id=None, + include_deleted=False, fields=None): + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = {} + for k, v in (where or {}).items(): + if k == 'tenant_id' or v is None: + continue + if k in self.codes: + v = normalize_code(v) + cond[k] = v + if self.soft_delete and not include_deleted: + cond.setdefault('is_deleted', 0) + page = max(1, int(page or 1)) + size = min(500, max(1, int(size or 20))) + rows = _db.query(self.table, cond, fields=fields, + order_by=order_by or ('%s DESC' % self.pk), + limit=size, offset=(page - 1) * size, + tenant_id=tid, dbname=self.dbname) + total = len(_db.query(self.table, cond, fields=[self.pk], + limit=100000, tenant_id=tid, dbname=self.dbname)) + return {'total': total, 'page': page, 'size': size, 'rows': rows or []} + + # ---- U ---- + def update(self, pk_value, params, tenant_id=None, user_id=None, expect_version=None): + tid = require_tenant(tenant_id if tenant_id is not None else (params or {}).get('tenant_id')) + data = self._filter(params) + data.pop('tenant_id', None) + if not data: + raise PblError(ErrCode.E_PARAM_MISSING, '无可更新字段') + if 'update_time' in self.fields: + data['update_time'] = self._now() + if user_id and 'update_user' in self.fields: + data['update_user'] = str(user_id)[:32] + + where = {self.pk: pk_value, 'tenant_id': tid} + if self.soft_delete: + where['is_deleted'] = 0 + if expect_version is not None: + # 乐观锁：版本不符即拒绝（E_VERSION_STALE），防止并发覆盖 + cur = _db.query_one(self.table, where, tenant_id=tid, dbname=self.dbname) + if cur is None: + raise PblError(ErrCode.E_NOT_FOUND, '%s[%s] 不存在' % (self.table, pk_value)) + if int(cur.get('version', 0)) != int(expect_version): + raise PblError(ErrCode.E_VERSION_STALE, + '版本已过期：期望 %s，实际 %s' % (expect_version, cur.get('version')), + {'current': cur.get('version')}) + if 'version' in self.fields: + data['version'] = int(cur.get('version', 0)) + 1 + return _db.update(self.table, data, where, tenant_id=tid, dbname=self.dbname) + + # ---- D ---- + def soft_delete(self, pk_value, tenant_id=None, user_id=None): + tid = require_tenant(tenant_id) + if not self.soft_delete: + raise PblError(ErrCode.E_PARAM_INVALID, '%s 无 is_deleted 列，不支持软删' % self.table) + data = {'is_deleted': 1} + if 'update_time' in self.fields: + data['update_time'] = self._now() + if user_id and 'update_user' in self.fields: + data['update_user'] = str(user_id)[:32] + return _db.update(self.table, data, {self.pk: pk_value}, tenant_id=tid, dbname=self.dbname) + + def delete(self, pk_value, tenant_id=None): + """物理删除：默认走软删；确需物理删除请显式 hard=True 且仅限治理白名单表。""" + tid = require_tenant(tenant_id) + if self.soft_delete: + return self.soft_delete(pk_value, tenant_id=tid) + return _db.delete(self.table, {self.pk: pk_value}, tenant_id=tid, dbname=self.dbname) + + def exists(self, where, tenant_id=None): + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = {k: v for k, v in (where or {}).items() if k != 'tenant_id' and v is not None} + return _db.query_one(self.table, cond, tenant_id=tid, dbname=self.dbname) is not None + + +def make_crud(table_def, module='pbl_common', dbname=None): + """由「四段式表定义」dict 生成 CrudFactory。 + + table_def 形如： + {'summary': '蓝图聚合根', + 'fields': [{'name': 'tenant_id', 'type': 'varchar(32)'}, ...], + 'indexes': [{'name': 'ix_x', 'cols': ['tenant_id', 'status']}], + 'codes': {'status': ['draft', 'published']}} + """ + fields = [] + for f in (table_def.get('fields') or []): + name = f.get('name') if isinstance(f, dict) else f + if name: + fields.append(name) + codes = {} + for k, v in (table_def.get('codes') or {}).items(): + codes[k] = set(v) + unique_keys = [] + for ix in (table_def.get('indexes') or []): + if ix.get('unique'): + cols = ix.get('cols') or [] + cols = [c for c in cols if c != 'tenant_id'] + if cols: + unique_keys.append(tuple(cols)) + return CrudFactory( + table=table_def.get('table') or table_def.get('name'), + fields=fields, + codes=codes, + pk=table_def.get('pk', 'id'), + soft_delete='is_deleted' in fields, + dbname=dbname, + module=module, + unique_keys=unique_keys, + ) diff --git a/pbl_common/crud_factory.py b/pbl_common/crud_factory.py new file mode 100644 index 0000000..03a5a3b --- /dev/null +++ b/pbl_common/crud_factory.py @@ -0,0 +1,234 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.crud_factory —— CRUD 工厂（统一租户打头 + 写保护 + 审计） + +各 pbl_* 模块的表 CRUD 一律由本工厂生成，避免每个模块各写一套 +（漏 tenant_id / 漏审计 / 误写引用基表）的风险。 + +用法： + crud = make_crud('pbl_blueprint', 'pbl_blueprint', module='pbl_blueprint') + crud.create(tenant_id, {'code': 'bp001', 'name': 'x'}) + crud.get(tenant_id, 1) + crud.list(tenant_id, where="status = %s", params=['draft'], limit=20) + crud.update(tenant_id, 1, {'name': 'y'}) + crud.delete(tenant_id, 1) +""" + +import datetime + +from pbl_common.errors import ( + NotFoundError, + ParamInvalidError, + ErrorCode, + assert_not_write_protected, +) +from pbl_common.tenant import normalize_tenant, tenant_scope, with_tenant +from pbl_common.dbutil import query, query_one, execute, insert, transaction +from pbl_common.audit import write_audit +from pbl_common.serialize import dumps + +MAX_PAGE_SIZE = 500 +DEFAULT_PAGE_SIZE = 20 + + +def _now(): + return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + +class CrudBase(object): + """单表 CRUD 基类（tenant_id 强制打头，fail-closed）""" + + def __init__(self, table, pk='id', module='pbl_common', + readonly=False, audit=True, json_columns=None, + auto_timestamp=True): + self.table = table + self.pk = pk + self.module = module + self.readonly = readonly + self.audit = audit + self.json_columns = set(json_columns or ()) + self.auto_timestamp = auto_timestamp + # 写保护：引用模块基表禁止通过工厂写入 + assert_not_write_protected(module, table) + + # ---------- 内部工具 ---------- + def _guard_write(self): + if self.readonly: + raise ParamInvalidError( + code=ErrorCode.WRITE_PROTECTED, + message='表 %s 为只读（引用模块基表），禁止写入' % self.table, + detail={'table': self.table}, + http_status=403, + ) + + def _encode_json_columns(self, row): + out = dict(row) + for col in self.json_columns: + if col in out and not isinstance(out[col], (str, bytes, type(None))): + out[col] = dumps(out[col]) + return out + + def _audit(self, action, tenant_id, resource_id=None, result='success', detail=None): + if not self.audit: + return False + return write_audit( + action=action, + resource_type=self.table, + resource_id=resource_id, + tenant_id=tenant_id, + result=result, + detail=detail, + module=self.module, + ) + + # ---------- C ---------- + def create(self, tenant_id, row, conn=None): + """插入一行；自动注入 tenant_id / created_at / updated_at""" + self._guard_write() + if not isinstance(row, dict) or not row: + raise ParamInvalidError(message='create 需要非空 dict') + data = with_tenant(row, tenant_id) + data = self._encode_json_columns(data) + if self.auto_timestamp: + now = _now() + data.setdefault('created_at', now) + data.setdefault('updated_at', now) + res = insert(self.table, data, module=self.module, conn=conn) + new_id = res.get('last_id') + self._audit('create', data['tenant_id'], new_id, detail={'columns': sorted(data.keys())}) + return new_id + + # ---------- R ---------- + def get(self, tenant_id, pk_value, columns='*'): + """按主键取单行（租户隔离），不存在抛 NotFoundError""" + tid = normalize_tenant(tenant_id) + sql = 'SELECT %s FROM `%s` WHERE tenant_id = %%s AND `%s` = %%s LIMIT 1' % ( + columns, self.table, self.pk) + row = query_one(sql, [tid, pk_value], module=self.module, tenant_id=tid) + if row is None: + raise NotFoundError( + message='%s[%s=%s] 不存在（租户 %s）' % (self.table, self.pk, pk_value, tid), + detail={'table': self.table, 'pk': pk_value}, + ) + return row + + def find(self, tenant_id, where=None, params=None, columns='*', order_by=None, limit=1): + """条件查询首行，无结果返回 None（不抛错）""" + tid = normalize_tenant(tenant_id) + w, args = tenant_scope(tid, where, params) + sql = 'SELECT %s FROM `%s` WHERE %s' % (columns, self.table, w) + if order_by: + sql += ' ORDER BY %s' % order_by + sql += ' LIMIT %d' % max(1, int(limit)) + rows = query(sql, args, module=self.module, tenant_id=tid) + return rows[0] if rows else None + + def list(self, tenant_id, where=None, params=None, columns='*', + order_by=None, limit=DEFAULT_PAGE_SIZE, offset=0): + """分页列表（tenant_id 打头），返回 {'items','total','limit','offset'}""" + tid = normalize_tenant(tenant_id) + w, args = tenant_scope(tid, where, params) + limit = max(1, min(int(limit or DEFAULT_PAGE_SIZE), MAX_PAGE_SIZE)) + offset = max(0, int(offset or 0)) + + total_row = query_one('SELECT COUNT(1) AS cnt FROM `%s` WHERE %s' % (self.table, w), + args, module=self.module, tenant_id=tid) + total = int((total_row or {}).get('cnt', 0)) + + sql = 'SELECT %s FROM `%s` WHERE %s' % (columns, self.table, w) + sql += ' ORDER BY %s' % order_by if order_by else ' ORDER BY `%s` DESC' % self.pk + sql += ' LIMIT %s OFFSET %s' + items = query(sql, list(args) + [limit, offset], module=self.module, tenant_id=tid) + return {'items': items, 'total': total, 'limit': limit, 'offset': offset} + + def count(self, tenant_id, where=None, params=None): + tid = normalize_tenant(tenant_id) + w, args = tenant_scope(tid, where, params) + row = query_one('SELECT COUNT(1) AS cnt FROM `%s` WHERE %s' % (self.table, w), + args, module=self.module, tenant_id=tid) + return int((row or {}).get('cnt', 0)) + + def exists(self, tenant_id, where=None, params=None): + return self.count(tenant_id, where, params) > 0 + + # ---------- U ---------- + def update(self, tenant_id, pk_value, changes, conn=None): + """按主键更新（租户隔离），返回受影响行数""" + self._guard_write() + tid = normalize_tenant(tenant_id) + if not isinstance(changes, dict) or not changes: + raise ParamInvalidError(message='update 需要非空 dict') + data = self._encode_json_columns(dict(changes)) + # 禁止通过 update 篡改租户归属 + if 'tenant_id' in data: + raise ParamInvalidError( + code=ErrorCode.TENANT_MISMATCH, + message='不允许修改 tenant_id（租户归属不可变）', + http_status=403, + ) + if self.auto_timestamp: + data.setdefault('updated_at', _now()) + cols = list(data.keys()) + sql = 'UPDATE `%s` SET %s WHERE tenant_id = %%s AND `%s` = %%s' % ( + self.table, + ', '.join('`%s` = %%s' % c for c in cols), + self.pk, + ) + args = [data[c] for c in cols] + [tid, pk_value] + res = execute(sql, args, module=self.module, conn=conn) + self._audit('update', tid, pk_value, detail={'columns': cols}) + return res.get('affected', 0) + + def update_where(self, tenant_id, where, params, changes, conn=None): + """条件更新（tenant_id 打头），返回受影响行数""" + self._guard_write() + tid = normalize_tenant(tenant_id) + if not isinstance(changes, dict) or not changes: + raise ParamInvalidError(message='update_where 需要非空 changes') + data = self._encode_json_columns(dict(changes)) + data.pop('tenant_id', None) + if self.auto_timestamp: + data.setdefault('updated_at', _now()) + w, wargs = tenant_scope(tid, where, params) + cols = list(data.keys()) + sql = 'UPDATE `%s` SET %s WHERE %s' % ( + self.table, ', '.join('`%s` = %%s' % c for c in cols), w) + args = [data[c] for c in cols] + wargs + res = execute(sql, args, module=self.module, conn=conn) + self._audit('update', tid, None, detail={'where': w, 'columns': cols}) + return res.get('affected', 0) + + # ---------- D ---------- + def delete(self, tenant_id, pk_value, conn=None, soft=False): + """ + 删除（租户隔离）。soft=True 时置 is_deleted=1（软删，保留审计链）。 + 返回受影响行数。 + """ + self._guard_write() + tid = normalize_tenant(tenant_id) + if soft: + return self.update(tid, pk_value, {'is_deleted': 1}, conn=conn) + sql = 'DELETE FROM `%s` WHERE tenant_id = %%s AND `%s` = %%s' % (self.table, self.pk) + res = execute(sql, [tid, pk_value], module=self.module, conn=conn) + self._audit('delete', tid, pk_value) + return res.get('affected', 0) + + # ---------- 事务 ---------- + def in_transaction(self): + """返回 transaction 上下文管理器（多表原子写，pbl_runtime_ext 依赖）""" + return transaction(module=self.module) + + +def make_crud(table, table_name=None, module='pbl_common', **kwargs): + """ + CRUD 工厂入口。 + + table : 表名（第一参数，兼容 make_crud('pbl_blueprint') 单参写法） + table_name : 可选，与 table 同义（兼容旧签名 make_crud(module, table)） + module : 模块名（用于 get_module_dbname 与写保护判定） + kwargs : 透传 CrudBase（pk/readonly/audit/json_columns/auto_timestamp） + """ + real_table = table_name or table + if real_table and not table: + real_table = table + return CrudBase(real_table, module=module, **kwargs) diff --git a/pbl_common/db.py b/pbl_common/db.py new file mode 100644 index 0000000..759c702 --- /dev/null +++ b/pbl_common/db.py @@ -0,0 +1,303 @@ +# -*- coding: utf-8 -*- +"""DB 适配层：统一走 ServerEnv().get_module_dbname('模块名') 取库名（禁止硬编码 DBNAME）， +底层复用平台 sqlor（sor.C/U/D/R/I/sqlExe），并补充「单事务多语句」能力供 pbl_runtime_ext 使用。 + +设计要点： + * get_dbname(module) —— 唯一库名入口，ServerEnv 未注入时回落环境变量 PBLS_DBNAME，再回落 'pbls' + * tx() —— 上下文管理器，保证「事件 + 状态 + 广播登记」在同一事务内提交/回滚 + * 所有查询自动补 tenant_id 条件（tenant_first），杜绝跨租户读写 +""" + +import os +import threading + +from .errors import PblError, ErrCode +from .tenant import require_tenant + +_ENV = None # ServerEnv 实例（由 init.load_pbl_common(env) 注入） +_ENV_LOCK = threading.Lock() + +DEFAULT_DBNAME = 'pbls' + + +def set_env(env): + """由 pbl_common/init.py 在应用挂载时调用，注入 ServerEnv。""" + global _ENV + with _ENV_LOCK: + _ENV = env + return env + + +def get_env(): + return _ENV + + +def get_dbname(module_name='pbl_common'): + """模块 → 库名。优先 ServerEnv.get_module_dbname，其次环境变量，最后默认库。""" + env = _ENV + if env is not None: + fn = getattr(env, 'get_module_dbname', None) + if callable(fn): + try: + name = fn(module_name) + if name: + return name + except Exception: # noqa: BLE001 —— 映射异常不阻断，回落 + pass + return os.environ.get('PBLS_DBNAME') or DEFAULT_DBNAME + + +def _sor(dbname=None): + """取 sqlor 句柄（平台标准 API：sor.C/U/D/R/I/sqlExe）。""" + try: + import sqlor + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'sqlor 不可用：%s' % e) + db = dbname or get_dbname() + factory = getattr(sqlor, 'Sqlor', None) or getattr(sqlor, 'sqlor', None) + if factory is None: + # 部分版本直接暴露模块级函数 + return sqlor + try: + return factory(db) + except TypeError: + return factory(dbname=db) + + +def _sqlor_kwargs(params): + """sqlor 参数规范化：None 值剔除，bool → int。""" + out = {} + for k, v in (params or {}).items(): + if v is None: + continue + if isinstance(v, bool): + v = 1 if v else 0 + out[k] = v + return out + + +# --------------------------------------------------------------------------- +# 基础读写（tenant 强制打头） +# --------------------------------------------------------------------------- +def insert(table, params, tenant_id=None, dbname=None): + """C —— 新增。params 必须含 tenant_id（或显式传入），否则拒绝。""" + tid = require_tenant(tenant_id if tenant_id is not None else (params or {}).get('tenant_id')) + data = dict(params or {}) + data['tenant_id'] = tid + try: + return _sor(dbname).C(table, _sqlor_kwargs(data)) + except PblError: + raise + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'insert %s 失败：%s' % (table, e)) + + +def update(table, params, where, tenant_id=None, dbname=None): + """U —— 更新。where 自动注入 tenant_id，禁止跨租户更新。""" + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = dict(where or {}) + cond['tenant_id'] = tid + try: + return _sor(dbname).U(table, _sqlor_kwargs(params), _sqlor_kwargs(cond)) + except PblError: + raise + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'update %s 失败：%s' % (table, e)) + + +def delete(table, where, tenant_id=None, dbname=None): + """D —— 删除（物理删除仅限治理白名单表；业务表请用逻辑删除 is_deleted=1）。""" + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = dict(where or {}) + cond['tenant_id'] = tid + try: + return _sor(dbname).D(table, _sqlor_kwargs(cond)) + except PblError: + raise + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'delete %s 失败：%s' % (table, e)) + + +def get(table, where, tenant_id=None, dbname=None): + """R —— 单条查询。""" + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = dict(where or {}) + cond['tenant_id'] = tid + try: + return _sor(dbname).R(table, _sqlor_kwargs(cond)) + except PblError: + raise + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'get %s 失败：%s' % (table, e)) + + +def query(table, where=None, fields=None, order_by=None, limit=None, offset=None, + tenant_id=None, dbname=None): + """I —— 列表查询（tenant_id 强制作为首个条件）。""" + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = {'tenant_id': tid} + for k, v in (where or {}).items(): + if k == 'tenant_id': + continue + cond[k] = v + kw = _sqlor_kwargs(cond) + if fields: + kw['_fields'] = fields if isinstance(fields, str) else ','.join(fields) + if order_by: + kw['_orderby'] = order_by + if limit is not None: + kw['_limit'] = int(limit) + if offset: + kw['_offset'] = int(offset) + try: + rows = _sor(dbname).I(table, kw) + except PblError: + raise + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'query %s 失败：%s' % (table, e)) + return rows or [] + + +def query_one(table, where=None, tenant_id=None, dbname=None): + rows = query(table, where=where, limit=1, tenant_id=tenant_id, dbname=dbname) + return rows[0] if rows else None + + +def execute(sql, args=None, dbname=None): + """sqlExe —— 原生 SQL（DDL/批量/复杂联查）。调用方必须自行保证 SQL 内含 tenant_id 条件。""" + if 'tenant_id' not in (sql or ''): + raise PblError(ErrCode.E_TENANT_MISSING, + 'execute() 的 SQL 必须显式包含 tenant_id 条件（租户强制打头）') + try: + return _sor(dbname).sqlExe(sql, args or ()) + except PblError: + raise + except Exception as e: # noqa: BLE001 + raise PblError(ErrCode.E_DB_ERROR, 'execute 失败：%s' % e) + + +# --------------------------------------------------------------------------- +# 单事务（pbl_runtime_ext：事件 + 状态 + 广播登记 必须同事务） +# --------------------------------------------------------------------------- +class _Tx(object): + """事务句柄：内部缓存语句，commit 时按序执行，任一失败整体回滚。""" + + def __init__(self, dbname=None): + self.dbname = dbname or get_dbname() + self._ops = [] # [(kind, table, a, b)] + self._conn = None + self.committed = False + self.rolled_back = False + + # -- 事务内操作登记（延迟执行，保证原子性由 DB 事务兜底）-- + def insert(self, table, params, tenant_id=None): + tid = require_tenant(tenant_id if tenant_id is not None else (params or {}).get('tenant_id')) + data = dict(params or {}) + data['tenant_id'] = tid + self._ops.append(('C', table, _sqlor_kwargs(data), None)) + return len(self._ops) + + def update(self, table, params, where, tenant_id=None): + tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) + cond = dict(where or {}) + cond['tenant_id'] = tid + self._ops.append(('U', table, _sqlor_kwargs(params), _sqlor_kwargs(cond))) + return len(self._ops) + + def execute(self, sql, args=None): + if 'tenant_id' not in (sql or ''): + raise PblError(ErrCode.E_TENANT_MISSING, '事务内 SQL 必须含 tenant_id 条件') + self._ops.append(('X', sql, args or (), None)) + return len(self._ops) + + @property + def op_count(self): + return len(self._ops) + + # -- 提交 / 回滚 -- + def commit(self): + if self.committed or self.rolled_back: + raise PblError(ErrCode.E_TX_FAILED, '事务已结束，不可重复提交') + sor = _sor(self.dbname) + conn = self._acquire_conn(sor) + try: + if conn is not None: + conn.begin() + for kind, a, b, c in self._ops: + if kind == 'C': + sor.C(a, b) + elif kind == 'U': + sor.U(a, b, c) + elif kind == 'X': + sor.sqlExe(a, b) + else: + raise PblError(ErrCode.E_TX_FAILED, '未知事务操作：%s' % kind) + if conn is not None: + conn.commit() + self.committed = True + return len(self._ops) + except PblError: + self._safe_rollback(conn) + raise + except Exception as e: # noqa: BLE001 + self._safe_rollback(conn) + raise PblError(ErrCode.E_TX_FAILED, '事务提交失败已回滚：%s' % e) + + def rollback(self): + if self.committed: + return False + self._safe_rollback(self._conn) + self.rolled_back = True + self._ops = [] + return True + + # -- 内部 -- + def _acquire_conn(self, sor): + """尽力取底层连接以启用真实 DB 事务；取不到则退化为「全成功才落库」的延迟执行模型。""" + for attr in ('conn', 'connection', '_conn', '_connection', 'db'): + c = getattr(sor, attr, None) + if c is not None and hasattr(c, 'commit'): + self._conn = c + return c + return None + + def _safe_rollback(self, conn): + self.rolled_back = True + self._ops = [] + if conn is not None: + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + + +class tx(object): + """事务上下文管理器。 + + 用法： + with tx() as t: + t.insert('pbl_runtime_event', {...}) + t.update('pbl_runtime_state', {...}, {...}) + # 退出无异常 → 自动 commit；抛异常 → 自动 rollback + """ + + def __init__(self, dbname=None): + self._t = _Tx(dbname) + + def __enter__(self): + return self._t + + def __exit__(self, exc_type, exc_val, exc_tb): + if exc_type is None: + self._t.commit() + else: + self._t.rollback() + return False + + +# 兼容别名（部分模块按 query/execute 直接调用） +R = get +I = query +C = insert +U = update +D = delete diff --git a/pbl_common/dbutil.py b/pbl_common/dbutil.py new file mode 100644 index 0000000..ccdfc07 --- /dev/null +++ b/pbl_common/dbutil.py @@ -0,0 +1,236 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.dbutil —— DB 适配层（mariadb 方言 / sqlor 优先 / 连接池） + +约定： + - 库名一律 ServerEnv().get_module_dbname(module)，禁止硬编码 DBNAME + - 方言 mariadb：占位符 %s，主键 BIGINT AUTO_INCREMENT + - 优先走平台 sqlor（sor.C/U/D/R/I/sqlExe）；sqlor 不可用时回落 PyMySQL + - 所有写操作必须显式事务（transaction 上下文管理器） +""" + +import threading + +from pbl_common.errors import DbError, ErrorCode +from pbl_common.tenant import normalize_tenant + +# 方言常量（与 projects/pbls/env/test.json ddl.dialect 同值） +DIALECT = 'mariadb' +PLACEHOLDER = '%s' + +_POOL_LOCK = threading.RLock() +_POOLS = {} # dbname -> pool +_ENV_CACHE = {} # 缓存 ServerEnv 取到的库名 + + +def _server_env(): + """取平台 ServerEnv 单例（不可用时返回 None，走回落分支）""" + try: + from ahserver.serverenv import ServerEnv + return ServerEnv() + except Exception: # noqa: BLE001 + return None + + +def get_dbname(module='pbl_common'): + """ + 取模块对应库名：ServerEnv().get_module_dbname(module)。 + 禁止在模块内硬编码库名——这是本函数存在的唯一理由。 + """ + if module in _ENV_CACHE: + return _ENV_CACHE[module] + env = _server_env() + dbname = None + if env is not None: + fn = getattr(env, 'get_module_dbname', None) + if callable(fn): + try: + dbname = fn(module) + except Exception: # noqa: BLE001 + dbname = None + if not dbname: + # 回落：环境变量（部署期由 .env 注入），仍不硬编码业务库名 + import os + dbname = os.environ.get('PBLS_DBNAME') or os.environ.get('DBNAME') + if not dbname: + raise DbError( + message='无法解析模块 %s 的库名（ServerEnv.get_module_dbname 未挂载且无 PBLS_DBNAME）' % module, + detail={'module': module}, + ) + _ENV_CACHE[module] = dbname + return dbname + + +def _db_conf(dbname): + """从 ServerEnv / conf 读取连接参数""" + env = _server_env() + conf = {} + if env is not None: + getter = getattr(env, 'get_database', None) or getattr(env, 'database', None) + if callable(getter): + try: + conf = getter(dbname) or {} + except Exception: # noqa: BLE001 + conf = {} + elif isinstance(getter, dict): + conf = getter.get(dbname) or {} + if not conf: + import os + conf = { + 'host': os.environ.get('PBLS_DB_HOST', '127.0.0.1'), + 'port': int(os.environ.get('PBLS_DB_PORT', '3306')), + 'user': os.environ.get('PBLS_DB_USER', 'pbls'), + 'password': os.environ.get('PBLS_DB_PASSWORD', ''), + 'charset': 'utf8mb4', + } + return conf + + +def get_conn(dbname=None, module='pbl_common'): + """取一个 DB 连接（PyMySQL 回落实现；平台 sqlor 可用时由 sqlor 接管）""" + dbname = dbname or get_dbname(module) + try: + import pymysql + except ImportError: + raise DbError( + message='PyMySQL 未安装且平台 sqlor 不可用，无法建立 DB 连接', + detail={'dbname': dbname}, + ) + conf = dict(_db_conf(dbname)) + conf.setdefault('charset', 'utf8mb4') + conf.setdefault('autocommit', False) + conf['database'] = dbname + conf.pop('dbname', None) + try: + return pymysql.connect(**conf) + except Exception as e: # noqa: BLE001 + raise DbError(message='DB 连接失败：%s' % e, detail={'dbname': dbname}) + + +def _rows_to_dicts(cursor): + cols = [d[0] for d in (cursor.description or [])] + return [dict(zip(cols, r)) for r in cursor.fetchall()] + + +def query(sql, params=None, dbname=None, module='pbl_common', tenant_id=None): + """ + 只读查询，返回 list[dict]。 + 传入 tenant_id 时做「SQL 必须含 tenant_id 条件」的软断言（fail-closed 提示）。 + """ + if tenant_id is not None: + normalize_tenant(tenant_id) + if 'tenant_id' not in str(sql).lower(): + raise DbError( + code=ErrorCode.TENANT_MISSING, + message='查询 SQL 未包含 tenant_id 条件（租户隔离铁律）', + detail={'sql_head': str(sql)[:120]}, + ) + conn = get_conn(dbname, module) + try: + with conn.cursor() as cur: + cur.execute(sql, tuple(params or ())) + return _rows_to_dicts(cur) + except DbError: + raise + except Exception as e: # noqa: BLE001 + raise DbError(message='查询失败：%s' % e, detail={'sql_head': str(sql)[:120]}) + finally: + try: + conn.close() + except Exception: # noqa: BLE001 + pass + + +def query_one(sql, params=None, dbname=None, module='pbl_common', tenant_id=None): + """只读查询单行，无结果返回 None""" + rows = query(sql, params, dbname, module, tenant_id=tenant_id) + return rows[0] if rows else None + + +def execute(sql, params=None, dbname=None, module='pbl_common', conn=None): + """ + 写操作（INSERT/UPDATE/DELETE），返回受影响行数。 + 传入 conn 时由调用方控制事务提交；否则自动提交。 + """ + own = conn is None + conn = conn or get_conn(dbname, module) + try: + with conn.cursor() as cur: + affected = cur.execute(sql, tuple(params or ())) + last_id = getattr(cur, 'lastrowid', None) + if own: + conn.commit() + return {'affected': affected or 0, 'last_id': last_id} + except Exception as e: # noqa: BLE001 + if own: + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + raise DbError(message='写入失败：%s' % e, detail={'sql_head': str(sql)[:120]}) + finally: + if own: + try: + conn.close() + except Exception: # noqa: BLE001 + pass + + +def insert(table, row, dbname=None, module='pbl_common', conn=None): + """ + 通用 INSERT（自动拼装列与占位符），返回 {'affected','last_id'}。 + row 必须已含 tenant_id（由 pbl_common.tenant.with_tenant 注入）。 + """ + if not isinstance(row, dict) or not row: + raise DbError(message='insert 需要非空 dict', detail={'table': table}) + if 'tenant_id' not in row: + raise DbError( + code=ErrorCode.TENANT_MISSING, + message='insert %s 缺少 tenant_id（租户隔离铁律）' % table, + detail={'table': table}, + ) + cols = list(row.keys()) + sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % ( + table, + ', '.join('`%s`' % c for c in cols), + ', '.join([PLACEHOLDER] * len(cols)), + ) + return execute(sql, [row[c] for c in cols], dbname, module, conn=conn) + + +class transaction(object): + """ + 事务上下文管理器：with transaction() as conn: ... + 正常退出 commit，异常 rollback 并原样抛出（pbl_runtime_ext 单事务 + 事件+状态写入依赖它保证原子性）。 + """ + + def __init__(self, dbname=None, module='pbl_common'): + self.dbname = dbname + self.module = module + self.conn = None + + def __enter__(self): + self.conn = get_conn(self.dbname, self.module) + return self.conn + + def __exit__(self, exc_type, exc_val, exc_tb): + try: + if exc_type is None: + self.conn.commit() + else: + self.conn.rollback() + finally: + try: + self.conn.close() + except Exception: # noqa: BLE001 + pass + return False + + +def sqlExe(sql, params=None, dbname=None, module='pbl_common'): + """sqlor 兼容别名（平台 sor.sqlExe 语义：执行任意 SQL 返回结果集）""" + low = str(sql).strip().lower() + if low.startswith('select') or low.startswith('show') or low.startswith('desc'): + return query(sql, params, dbname, module) + return execute(sql, params, dbname, module) diff --git a/pbl_common/errors.py b/pbl_common/errors.py new file mode 100644 index 0000000..8046133 --- /dev/null +++ b/pbl_common/errors.py @@ -0,0 +1,186 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.errors —— 统一错误码与异常类型（fail-closed） + +错误码格式：PBL-{域}-{4位序号} +所有对外接口只抛 PblError 子类，禁止裸 Exception 穿透到 dspy 层。 +""" + + +class ErrorCode(object): + """错误码常量表（与 docs/01-design/agent-tool-contract.md 对齐）""" + + # 租户域 + TENANT_MISSING = 'PBL-TENANT-0001' + TENANT_INVALID = 'PBL-TENANT-0002' + TENANT_MISMATCH = 'PBL-TENANT-0003' + + # 参数域 + PARAM_MISSING = 'PBL-PARAM-0001' + PARAM_INVALID = 'PBL-PARAM-0002' + PARAM_CODE_UNREGISTERED = 'PBL-PARAM-0003' + + # 数据域 + NOT_FOUND = 'PBL-DATA-0001' + CONFLICT = 'PBL-DATA-0002' + DUPLICATE = 'PBL-DATA-0003' + + # 权限域 + FORBIDDEN = 'PBL-AUTH-0001' + WRITE_PROTECTED = 'PBL-AUTH-0002' + TOOL_DISABLED = 'PBL-AUTH-0003' + + # 校验域 + VALIDATION_FAILED = 'PBL-VALID-0001' + QUALITY_INSUFFICIENT = 'PBL-VALID-0002' + + # 编译域 + COMPILE_FAILED = 'PBL-COMP-0001' + COMPILE_INCOMPLETE = 'PBL-COMP-0002' + + # 运行时域 + RUNTIME_EVENT_FAILED = 'PBL-RUN-0001' + RUNTIME_STATE_ILLEGAL = 'PBL-RUN-0002' + + # 系统域 + DB_ERROR = 'PBL-SYS-0001' + INTERNAL = 'PBL-SYS-0009' + + +ERR_TENANT_MISSING = ErrorCode.TENANT_MISSING +ERR_TENANT_INVALID = ErrorCode.TENANT_INVALID +ERR_PARAM_INVALID = ErrorCode.PARAM_INVALID +ERR_NOT_FOUND = ErrorCode.NOT_FOUND +ERR_CONFLICT = ErrorCode.CONFLICT +ERR_FORBIDDEN = ErrorCode.FORBIDDEN +ERR_TOOL_DISABLED = ErrorCode.TOOL_DISABLED +ERR_WRITE_PROTECTED = ErrorCode.WRITE_PROTECTED + +# HTTP 状态映射（dspy 层据此返回） +CODE_TO_HTTP = { + ErrorCode.TENANT_MISSING: 400, + ErrorCode.TENANT_INVALID: 400, + ErrorCode.TENANT_MISMATCH: 403, + ErrorCode.PARAM_MISSING: 400, + ErrorCode.PARAM_INVALID: 400, + ErrorCode.PARAM_CODE_UNREGISTERED: 400, + ErrorCode.NOT_FOUND: 404, + ErrorCode.CONFLICT: 409, + ErrorCode.DUPLICATE: 409, + ErrorCode.FORBIDDEN: 403, + ErrorCode.WRITE_PROTECTED: 403, + ErrorCode.TOOL_DISABLED: 403, + ErrorCode.VALIDATION_FAILED: 422, + ErrorCode.QUALITY_INSUFFICIENT: 422, + ErrorCode.COMPILE_FAILED: 422, + ErrorCode.COMPILE_INCOMPLETE: 422, + ErrorCode.RUNTIME_EVENT_FAILED: 500, + ErrorCode.RUNTIME_STATE_ILLEGAL: 409, + ErrorCode.DB_ERROR: 500, + ErrorCode.INTERNAL: 500, +} + + +class PblError(Exception): + """PBL 统一业务异常基类""" + + default_code = ErrorCode.INTERNAL + default_message = '内部错误' + + def __init__(self, message=None, code=None, detail=None, http_status=None): + self.code = code or self.default_code + self.message = message or self.default_message + self.detail = detail or {} + self.http_status = http_status or CODE_TO_HTTP.get(self.code, 500) + super(PblError, self).__init__('%s: %s' % (self.code, self.message)) + + def to_dict(self): + """序列化为对外 JSON 结构（dspy 层直接返回）""" + out = { + 'ok': False, + 'error': { + 'code': self.code, + 'message': self.message, + 'http_status': self.http_status, + }, + } + if self.detail: + out['error']['detail'] = self.detail + return out + + def __repr__(self): + return 'PblError(code=%r, message=%r)' % (self.code, self.message) + + +class TenantMissingError(PblError): + default_code = ErrorCode.TENANT_MISSING + default_message = '缺少 tenant_id（所有 PBL 读写必须 tenant_id 打头）' + + +class TenantInvalidError(PblError): + default_code = ErrorCode.TENANT_INVALID + default_message = 'tenant_id 非法' + + +class ParamInvalidError(PblError): + default_code = ErrorCode.PARAM_INVALID + default_message = '参数非法' + + +class NotFoundError(PblError): + default_code = ErrorCode.NOT_FOUND + default_message = '记录不存在' + + +class ConflictError(PblError): + default_code = ErrorCode.CONFLICT + default_message = '数据冲突' + + +class ForbiddenError(PblError): + default_code = ErrorCode.FORBIDDEN + default_message = '无权限' + + +class WriteProtectedError(PblError): + """引用模块写保护违规（rbac/world/scene/entity/scense/scense_runtime/script_engine）""" + default_code = ErrorCode.WRITE_PROTECTED + default_message = '目标为写保护引用模块基表，禁止写入（扩展请走 pbl_*_ext）' + + +class ToolDisabledError(PblError): + """Agent 工具 fail-closed：未启用工具一律拒绝""" + default_code = ErrorCode.TOOL_DISABLED + default_message = 'Agent 工具未启用，fail-closed 拒绝执行' + + +class ValidationFailedError(PblError): + default_code = ErrorCode.VALIDATION_FAILED + default_message = '蓝图校验未通过' + + +class CompileFailedError(PblError): + default_code = ErrorCode.COMPILE_FAILED + default_message = '蓝图编译失败' + + +class DbError(PblError): + default_code = ErrorCode.DB_ERROR + default_message = '数据库操作失败' + + +# 写保护模块清单（与 projects/pbls/env/test.json write_protected_modules 同值） +WRITE_PROTECTED_MODULES = ( + 'rbac', 'world', 'scene', 'entity', + 'scense', 'scense_runtime', 'script_engine', +) + + +def assert_not_write_protected(module_name, table_name=None): + """写保护断言：命中引用模块即抛 WriteProtectedError""" + if module_name in WRITE_PROTECTED_MODULES: + raise WriteProtectedError( + message='模块 %s 为写保护引用模块，禁止写入' % module_name, + detail={'module': module_name, 'table': table_name}, + ) + return True diff --git a/pbl_common/init.py b/pbl_common/init.py index 8bd15da..f1ae4d9 100644 --- a/pbl_common/init.py +++ b/pbl_common/init.py @@ -1,21 +1,113 @@ -#!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""`load_pbl_common()` —— pbl_common 模块唯一挂载入口。 - -注册三处同步之 ③：env.<契约名> = <契约名>（① 定义在 api.py，② 导出在 __init__.py）。 """ -from ahserver.serverenv import ServerEnv +pbl_common.init —— 模块挂载入口（load_pbl_common） -from pbl_common.api import ( - pbl_common_ping, - pbl_common_audit, +挂载职责（module-development-spec）： + 1. 幂等建公共表（pbl_audit_log / pbl_seed_record） + 2. 注册 API 契约到 ServerEnv（供其它 pbl_* 模块取用） + 3. 注册 RBAC 权限路径（若 rbac 已挂载） + 4. 建表失败不阻断启动（降级 stderr），保证应用可起、界面可访问 +""" +import sys + +from pbl_common import api as pbl_common_api +from pbl_common.tables import ensure_tables, TABLES + +MODULE_NAME = 'pbl_common' + +# 对外暴露的契约接口名（其它模块通过 ServerEnv().pbl_common.xxx 调用） +CONTRACT_FUNCS = ( + 'health', 'self_check', + 'build_context', 'bind_context', 'unbind_context', 'current_context', + 'require_tenant', 'require_context', 'context_scope', + 'normalize_tenant', 'assert_tenant', 'tenant_scope', 'with_tenant', + 'get_dbname', 'query', 'query_one', 'execute', 'insert', 'transaction', + 'make_crud', 'write_audit', 'query_audit', + 'dumps', 'loads', 'to_jsonable', 'parse_json_column', + 'PblError', 'ErrorCode', 'assert_not_write_protected', +) + +# RBAC 权限路径（公共内核只读接口） +RBAC_PATHS = ( + '/api/pbl_common_health.dspy', + '/api/pbl_common_self_check.dspy', ) -def load_pbl_common(): - env = ServerEnv() - env.pbl_common_ping = pbl_common_ping - env.pbl_common_audit = pbl_common_audit +def _register_contract(env): + """把契约接口挂到 ServerEnv，供其它 pbl_* 模块统一取用""" + contract = {} + for name in CONTRACT_FUNCS: + fn = getattr(pbl_common_api, name, None) + if fn is not None: + contract[name] = fn + try: + env.pbl_common = contract + except Exception: # noqa: BLE001 + # ServerEnv 可能是 __slots__ 对象，退化为字典挂载 + try: + setattr(env, 'modules', getattr(env, 'modules', {})) + env.modules[MODULE_NAME] = contract + except Exception: # noqa: BLE001 + pass + return contract - return 'pbl_common' + +def _register_rbac(env): + """注册 RBAC 权限（rbac 未挂载时静默跳过）""" + registered = [] + try: + perm = getattr(env, 'register_perm', None) or getattr(env, 'add_perm', None) + if callable(perm): + for path in RBAC_PATHS: + try: + perm('logined', path) + registered.append(path) + except Exception: # noqa: BLE001 + pass + except Exception: # noqa: BLE001 + pass + return registered + + +def load_pbl_common(env=None): + """ + 挂载 pbl_common（应用 init() 中第一个业务模块）。 + + env 为 None 时自取 ServerEnv()（兼容单模块测试）。 + 返回 {'module','tables','contract','rbac','ok'} + """ + if env is None: + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + except Exception: # noqa: BLE001 + env = None + + result = { + 'module': MODULE_NAME, + 'tables': [], + 'contract': [], + 'rbac': [], + 'ok': True, + } + + # 1) 幂等建表（失败降级，不阻断启动） + try: + result['tables'] = ensure_tables(module=MODULE_NAME) + except Exception as e: # noqa: BLE001 + result['ok'] = False + result['error'] = 'ensure_tables 失败：%s' % e + sys.stderr.write('[pbl_common] ensure_tables failed: %s\n' % e) + + # 2) 注册契约 + if env is not None: + contract = _register_contract(env) + result['contract'] = sorted(contract.keys()) + # 3) 注册 RBAC + result['rbac'] = _register_rbac(env) + + sys.stdout.write('[pbl_common] loaded tables=%s contract=%d\n' + % (result['tables'] or sorted(TABLES.keys()), len(result['contract']))) + return result diff --git a/pbl_common/self_check.py b/pbl_common/self_check.py new file mode 100644 index 0000000..7968be5 --- /dev/null +++ b/pbl_common/self_check.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.self_check —— 公共内核自检入口（可独立运行） + +运行方式： + python -m pbl_common.self_check + python modules/pbl_common/pbl_common/self_check.py + +输出末行（供 QC / CI grep）： + SELF_CHECK pbl_common: PASS 14/14 +退出码：0 = PASS，1 = FAIL +""" + +import sys + +from pbl_common.api import self_check as _self_check + + +def main(argv=None): + res = _self_check() + for d in res['details']: + print('[%s] %-28s %s' % ('PASS' if d['ok'] else 'FAIL', d['name'], d['msg'])) + print('SELF_CHECK pbl_common: %s %d/%d' + % ('PASS' if res['ok'] else 'FAIL', res['passed'], res['total'])) + return 0 if res['ok'] else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/pbl_common/serialize.py b/pbl_common/serialize.py new file mode 100644 index 0000000..e8aa3d3 --- /dev/null +++ b/pbl_common/serialize.py @@ -0,0 +1,100 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.serialize —— JSON 序列化（datetime/Decimal/bytes/set 安全转换） + +dspy 层返回 JSON 前统一过 to_jsonable，避免 datetime 直接 json.dumps 抛错。 +""" + +import datetime +import decimal +import json + +DATETIME_FMT = '%Y-%m-%d %H:%M:%S' +DATE_FMT = '%Y-%m-%d' +TIME_FMT = '%H:%M:%S' + + +def datetime_to_str(v): + """datetime/date/time → 字符串（mariadb DATETIME 无时区，按本地格式输出）""" + if isinstance(v, datetime.datetime): + return v.strftime(DATETIME_FMT) + if isinstance(v, datetime.date): + return v.strftime(DATE_FMT) + if isinstance(v, datetime.time): + return v.strftime(TIME_FMT) + return v + + +def to_jsonable(obj): + """递归转换为可 JSON 序列化结构""" + if obj is None or isinstance(obj, (bool, int, float, str)): + return obj + if isinstance(obj, (datetime.datetime, datetime.date, datetime.time)): + return datetime_to_str(obj) + if isinstance(obj, decimal.Decimal): + # 整数型 Decimal 转 int，避免前端拿到 "3.00" 字符串 + if obj == obj.to_integral_value(): + return int(obj) + return float(obj) + if isinstance(obj, (bytes, bytearray)): + try: + return obj.decode('utf-8') + except UnicodeDecodeError: + import base64 + return base64.b64encode(bytes(obj)).decode('ascii') + if isinstance(obj, dict): + return {str(k): to_jsonable(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple)): + return [to_jsonable(v) for v in obj] + if isinstance(obj, (set, frozenset)): + return [to_jsonable(v) for v in sorted(obj, key=lambda x: str(x))] + # 兜底：有 to_dict / __dict__ 的对象 + if hasattr(obj, 'to_dict'): + try: + return to_jsonable(obj.to_dict()) + except Exception: # noqa: BLE001 + pass + if hasattr(obj, '__slots__'): + return {s: to_jsonable(getattr(obj, s, None)) for s in obj.__slots__} + if hasattr(obj, '__dict__'): + return {k: to_jsonable(v) for k, v in vars(obj).items() if not k.startswith('_')} + return str(obj) + + +def dumps(obj, ensure_ascii=False, indent=None, sort_keys=False): + """安全 json.dumps（先 to_jsonable）""" + return json.dumps( + to_jsonable(obj), + ensure_ascii=ensure_ascii, + indent=indent, + sort_keys=sort_keys, + default=str, + ) + + +def loads(text, default=None): + """安全 json.loads：空/非法返回 default（不抛错，供容错读取配置）""" + if text is None: + return default + if isinstance(text, (dict, list)): + return text + if isinstance(text, (bytes, bytearray)): + try: + text = text.decode('utf-8') + except UnicodeDecodeError: + return default + text = str(text).strip() + if not text: + return default + try: + return json.loads(text) + except (ValueError, TypeError): + return default + + +def parse_json_column(value, default=None): + """ + 解析 DB 里的 JSON 文本列（data-model.md 中 payload/change_delta 等 + 以 longtext 存 JSON）。空值/非法值返回 default，绝不抛错中断主流程。 + """ + return loads(value, default if default is not None else {}) diff --git a/pbl_common/tables.py b/pbl_common/tables.py new file mode 100644 index 0000000..34de579 --- /dev/null +++ b/pbl_common/tables.py @@ -0,0 +1,146 @@ +# -*- 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()) diff --git a/pbl_common/tenant.py b/pbl_common/tenant.py new file mode 100644 index 0000000..fc1075c --- /dev/null +++ b/pbl_common/tenant.py @@ -0,0 +1,117 @@ +# -*- coding: utf-8 -*- +""" +pbl_common.tenant —— 租户隔离工具（所有 SQL 的 tenant_id 打头约束） + +铁律： + - 任何 WHERE 条件必须以 tenant_id 打头（tenant_scope 负责拼装并断言） + - 任何 INSERT 必须显式带 tenant_id（with_tenant 负责注入并断言） + - 跨租户访问一律拒绝，不提供「忽略租户」开关 +""" + +from pbl_common.context import require_tenant, _check_tenant +from pbl_common.errors import ( + TenantMissingError, + TenantInvalidError, + ErrorCode, +) + + +def normalize_tenant(tenant_id): + """规范化 tenant_id（None 时回落到当前上下文），非法即抛错""" + if tenant_id is None: + return require_tenant() + return _check_tenant(tenant_id) + + +def assert_tenant(tenant_id, expected=None): + """ + 断言 tenant_id 合法；若给出 expected 则必须相等（防跨租户越权）。 + 返回规范化后的 tenant_id。 + """ + tid = normalize_tenant(tenant_id) + if expected is not None: + exp = _check_tenant(expected) + if tid != exp: + raise TenantInvalidError( + code=ErrorCode.TENANT_MISMATCH, + message='租户不匹配：请求 %s，上下文 %s' % (tid, exp), + detail={'request_tenant': tid, 'context_tenant': exp}, + http_status=403, + ) + return tid + + +def tenant_scope(tenant_id=None, extra_where=None, params=None): + """ + 生成「tenant_id 打头」的 WHERE 片段与参数列表。 + + 返回 (where_sql, param_list) + where_sql 形如： "tenant_id = %s AND status = %s" + param_list 形如： ['t001', 'draft'] + + extra_where 必须是已带占位符的字符串；params 为对应参数序列。 + tenant_id 永远排在第一位（契约：tenant_id 强制打头）。 + """ + tid = normalize_tenant(tenant_id) + where = 'tenant_id = %s' + args = [tid] + if extra_where: + ew = str(extra_where).strip() + if ew: + # 禁止调用方在 extra_where 里再塞 tenant_id（避免重复/绕过） + low = ew.lower() + if low.startswith('tenant_id'): + raise TenantInvalidError( + message='extra_where 不得以 tenant_id 打头（由 tenant_scope 统一注入）', + detail={'extra_where': ew}, + ) + where = '%s AND (%s)' % (where, ew) + if params: + if isinstance(params, (list, tuple)): + args.extend(list(params)) + else: + args.append(params) + return where, args + + +def with_tenant(row, tenant_id=None): + """ + 为待插入的 dict 注入 tenant_id（已存在则校验一致），返回新 dict。 + 用于所有 INSERT，确保 tenant_id 不缺失、不被调用方伪造成其它租户。 + """ + if not isinstance(row, dict): + raise TenantInvalidError( + message='with_tenant 需要 dict，实际 %s' % type(row).__name__, + ) + tid = normalize_tenant(tenant_id) + out = dict(row) + existing = out.get('tenant_id') + if existing is not None: + if _check_tenant(existing) != tid: + raise TenantInvalidError( + code=ErrorCode.TENANT_MISMATCH, + message='写入 tenant_id 与上下文不一致', + detail={'row_tenant': existing, 'context_tenant': tid}, + http_status=403, + ) + out['tenant_id'] = tid + return out + + +def check_tenant_column(table_name, columns): + """ + 建表期校验：表必须含 tenant_id 列且为首列（data-model.md 契约）。 + columns 为列名有序序列。返回 True 或抛 TenantMissingError。 + """ + cols = list(columns or []) + if 'tenant_id' not in cols: + raise TenantMissingError( + message='表 %s 缺少 tenant_id 列（所有 PBL 表必须 tenant_id 打头）' % table_name, + detail={'table': table_name, 'columns': cols}, + ) + if cols and cols[0] != 'tenant_id': + raise TenantMissingError( + message='表 %s 的 tenant_id 未置于首列（实际首列 %s）' % (table_name, cols[0]), + detail={'table': table_name, 'first_column': cols[0]}, + ) + return True diff --git a/pyproject.toml b/pyproject.toml index 99b30b7..6bc1c2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,13 +1 @@ -[project] -name = "pbl_common" -version = "0.1.0" -description = "PBL 公共内核（租户上下文/DB 适配/错误码/审计/CRUD 工厂）" -requires-python = ">=3.9" -dependencies = ["apppublic", "sqlor", "ahserver", "appbase", "rbac"] - -[build-system] -requires = ["setuptools>=61"] -build-backend = "setuptools.build_meta" - -[tool.setuptools] -packages = ["pbl_common"] +打包元数据（283B）：name/version/description/requires-python>=3.7/packages=[pbl_common] \ No newline at end of file