194 lines
7.7 KiB
Python
194 lines
7.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pbl_common.init —— load_pbl_common()：把公共内核注册到宿主 ServerEnv。
|
||
|
||
【QC 退回意见 #7 的修复】
|
||
① ServerEnv 来源二义：旧 init.py 写 ``from appbase.serverenv import ServerEnv``，
|
||
而同包 dbutil.py 写 ``ahserver.serverenv``。经核对 module-development-spec
|
||
（『ahserver ServerEnv』为基座）与 web-application-spec（应用入口
|
||
``from ahserver.webapp import webapp``），**正确来源是 ``ahserver.serverenv``**；
|
||
``appbase.serverenv`` 仅作历史回落候选。本版统一走 ``get_server_env()``
|
||
（dbutil 内实现，按 ahserver → appbase 顺序探测），init.py 不再自己写死 import，
|
||
两处不可能再互相矛盾。
|
||
② 旧自检完全依赖非本次交付的 kernel.py 符号体系 → 本版自检改为调用
|
||
``pbl_common.self_check.run_self_check()``（针对真实交付的
|
||
errors/context/dbutil/crud_factory/audit/api 编写断言）。
|
||
③ 包内两套平行实现（旧 kernel/tenant/crud/db/serialize vs 新交付
|
||
errors/context/crud_factory/dbutil）→ **保留新交付一套为唯一实现**；
|
||
旧模块名以「薄兼容 shim」形式存在（tenant.py 从 context 再导出、
|
||
crud.py 从 crud_factory 再导出、db.py 从 dbutil 再导出、serialize.py 独立小工具），
|
||
kernel.py 若存在则由其从本套再导出，方向恒为 kernel → 新实现，无环。
|
||
|
||
【三处同步（module-development-spec CRITICAL）】
|
||
① 实现：本文件定义 ``load_pbl_common``；
|
||
② ``pbl_common/__init__.py`` 导出 ``load_pbl_common``；
|
||
③ 宿主入口 ``apps/pbls/app/pbls.py`` 的 ``init()`` 里显式 ``load_pbl_common()``。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from pbl_common import api as _api
|
||
from pbl_common import audit as _audit
|
||
from pbl_common import context as _context
|
||
# 注意：包 __init__.py 把 crud_factory **函数**再导出到 pbl_common 命名空间，
|
||
# 会遮蔽同名子模块 → 这里必须用 importlib 取模块对象，不能写
|
||
# `from pbl_common import crud_factory as _crud`（会拿到函数而非模块）。
|
||
import importlib as _importlib
|
||
_crud = _importlib.import_module('pbl_common.crud_factory')
|
||
from pbl_common import dbutil as _db
|
||
from pbl_common import errors as _errors
|
||
|
||
log = logging.getLogger('pbl_common.init')
|
||
|
||
#: 注册到 ServerEnv 的符号清单（三处同步之第③处的注册面）
|
||
ENV_EXPORTS = {
|
||
# errors
|
||
'pbl_err': _errors.err,
|
||
'pbl_fail': _errors.fail,
|
||
'pbl_error_body': _errors.error_body,
|
||
'pbl_http_status_of': _errors.http_status_of,
|
||
'pbl_normalize_code': _errors.normalize_code,
|
||
'pbl_assert_not_write_protected': _errors.assert_not_write_protected,
|
||
'pbl_is_write_protected': _errors.is_write_protected,
|
||
'pbl_assert_append_only': _errors.assert_append_only,
|
||
# context
|
||
'pbl_normalize_tenant': _context.normalize_tenant,
|
||
'pbl_assert_tenant': _context.assert_tenant,
|
||
'pbl_tenant_scope': _context.tenant_scope,
|
||
'pbl_get_tenant': _context.get_tenant,
|
||
'pbl_set_tenant': _context.set_tenant,
|
||
'pbl_reset_tenant': _context.reset_tenant,
|
||
'pbl_current_context': _context.current_context,
|
||
'pbl_with_tenant': _context.with_tenant,
|
||
'pbl_check_tenant_column': _context.check_tenant_column,
|
||
# dbutil
|
||
'pbl_get_db': _db.get_db,
|
||
'pbl_new_id': _db.new_id,
|
||
'pbl_now_str': _db.now_str,
|
||
'pbl_sqlExe': _db.sqlExe,
|
||
'pbl_sqlor_available': _db.sqlor_available,
|
||
# crud
|
||
'pbl_tenant_crud': _crud.tenant_crud,
|
||
'pbl_crud_factory': _crud.crud_factory,
|
||
'pbl_bulk_create': _crud.bulk_create,
|
||
# audit
|
||
'pbl_write_audit': _audit.write_audit,
|
||
'pbl_write_audit_batch': _audit.write_audit_batch,
|
||
'pbl_audit_trail': _audit.audit_trail,
|
||
'pbl_flush_memory_audit': _audit.flush_memory_audit,
|
||
# api
|
||
'pbl_ok': _api.ok,
|
||
'pbl_fail_body': _api.fail_body,
|
||
'pbl_paged': _api.paged,
|
||
'pbl_api_guard': _api.api_guard,
|
||
}
|
||
|
||
_loaded = {'done': False, 'env': None, 'registered': 0}
|
||
|
||
|
||
def get_env():
|
||
"""取宿主 ServerEnv 单例（ahserver.serverenv 优先，appbase 回落）。"""
|
||
return _db.get_server_env()
|
||
|
||
|
||
def register_to_env(env=None, exports=None):
|
||
"""把 ENV_EXPORTS 挂到 ServerEnv。env 为 None 时自动探测；探测不到返回 0（不抛）。"""
|
||
env = env if env is not None else get_env()
|
||
if env is None:
|
||
log.warning('未探测到 ServerEnv（ahserver/appbase 均不可用），跳过注册')
|
||
return 0
|
||
table = exports or ENV_EXPORTS
|
||
count = 0
|
||
for name, func in table.items():
|
||
try:
|
||
setattr(env, name, func)
|
||
count += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
log.error('注册 %s 到 ServerEnv 失败：%s', name, exc)
|
||
return count
|
||
|
||
|
||
def load_pbl_common(env=None, run_check=False, verbose=False):
|
||
"""挂载公共内核（幂等）。
|
||
|
||
参数：
|
||
env —— 宿主 ServerEnv；None 时自动探测
|
||
run_check —— 挂载后执行 self_check（部署冒烟用）
|
||
verbose —— 打印自检明细
|
||
|
||
返回：{'loaded': True, 'registered': N, 'self_check': (passed, total) | None}
|
||
"""
|
||
registered = register_to_env(env)
|
||
_loaded.update({'done': True, 'env': env if env is not None else get_env(),
|
||
'registered': registered})
|
||
# 契约门禁：import 期已校验，这里再显式跑一次（半迁移状态禁止上线）
|
||
missing = _api.verify_contract()
|
||
if missing:
|
||
raise ImportError('pbl_common 契约符号缺失：%s' % ', '.join(missing))
|
||
result = {'loaded': True, 'registered': registered, 'self_check': None,
|
||
'sqlor': _db.sqlor_available(),
|
||
'fallback': _db.fallback_enabled()}
|
||
if run_check:
|
||
from pbl_common.self_check import run_self_check
|
||
passed, total, failures = run_self_check(verbose=verbose)
|
||
result['self_check'] = (passed, total)
|
||
if failures:
|
||
raise AssertionError('pbl_common 自检未通过 %d/%d：%s'
|
||
% (passed, total, failures[0][1]))
|
||
log.info('load_pbl_common 完成：registered=%d sqlor=%s', registered, result['sqlor'])
|
||
return result
|
||
|
||
|
||
def is_loaded():
|
||
return bool(_loaded['done'])
|
||
|
||
|
||
def load_status():
|
||
return dict(_loaded)
|
||
|
||
|
||
def unload_pbl_common(env=None):
|
||
"""卸载（仅测试用）：摘掉注册符号。"""
|
||
env = env if env is not None else _loaded.get('env')
|
||
removed = 0
|
||
if env is not None:
|
||
for name in ENV_EXPORTS:
|
||
if hasattr(env, name):
|
||
try:
|
||
delattr(env, name)
|
||
removed += 1
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
_loaded.update({'done': False, 'env': None, 'registered': 0})
|
||
return removed
|
||
|
||
|
||
# ==========================================================================
|
||
# 部署期自检入口（不依赖 kernel.py）
|
||
# ==========================================================================
|
||
def self_check(verbose=True):
|
||
"""执行 pbl_common 全量自检，返回 (passed, total, failures)。"""
|
||
from pbl_common.self_check import run_self_check
|
||
return run_self_check(verbose=verbose)
|
||
|
||
|
||
def main():
|
||
import sys
|
||
info = load_pbl_common(run_check=True, verbose=True)
|
||
print('LOAD pbl_common: registered=%d sqlor=%s self_check=%s'
|
||
% (info['registered'], info['sqlor'], info['self_check']))
|
||
passed, total = info['self_check'] or (0, 0)
|
||
return 0 if passed == total and total > 0 else 1
|
||
|
||
|
||
if __name__ == '__main__':
|
||
import sys
|
||
sys.exit(main())
|
||
|
||
|
||
__all__ = [
|
||
'ENV_EXPORTS', 'load_pbl_common', 'register_to_env', 'get_env',
|
||
'is_loaded', 'load_status', 'unload_pbl_common', 'self_check', 'main',
|
||
]
|