yumoqing/pccs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: apply monkey-patches at module level before any imports

Browse Source
This commit is contained in:
yumoqing 2026-08-11 16:57:13 +08:00
parent dd028e2203
commit 9d3c65a056
1 changed files with 38 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
app/pccs.py
View File

@ -6,23 +6,25 @@ import os, sys
from ahserver.webapp import webapp from ahserver.webapp import webapp
# ═══ 模块级 Monkey Patches (必须在任何 rbac/sqlor 使用前执行) ═══
def init(): # 1. DictObject.__getattr__ — 所有 ServerEnv 实例默认 get_module_dbname
# Monkey-patch DictObject.__getattr__ — 所有 ServerEnv 实例默认返回 'pccs'
from appPublic.dictObject import DictObject from appPublic.dictObject import DictObject
_orig = DictObject.__getattr__ _orig_getattr = DictObject.__getattr__
def _patched(self, key):
def _patched_getattr(self, key):
if key == 'get_module_dbname': if key == 'get_module_dbname':
return lambda m: 'pccs' return lambda m: 'pccs'
if key == 'password_encode': if key == 'password_encode':
return lambda s: s return lambda s: s
return _orig(self, key) return _orig_getattr(self, key)
DictObject.__getattr__ = _patched
# Monkey-patch get_sor_context — 复用已初始化的 DBPools 而非创建空实例 DictObject.__getattr__ = _patched_getattr
# 2. get_sor_context — 复用全局 DBPools 而非创建空实例
from appPublic.jsonConfig import getConfig from appPublic.jsonConfig import getConfig
import sqlor.dbpools as _dbp import sqlor.dbpools as _dbp
from sqlor.dbpools import DBPools from sqlor.dbpools import DBPools
@ -32,13 +34,18 @@ def init():
_orig_sor_ctx = _dbp.get_sor_context _orig_sor_ctx = _dbp.get_sor_context
async def _patched_sor_ctx(env, modulename, errorback=None): async def _patched_sor_ctx(env, modulename, errorback=None):
dbname = env.get_module_dbname(modulename) dbname = env.get_module_dbname(modulename)
async with _global_db.sqlorContext(dbname) as sor: async with _global_db.sqlorContext(dbname) as sor:
yield sor yield sor
_dbp.get_sor_context = _patched_sor_ctx
_dbp.get_sor_context = _patched_sor_ctx
_db_patched = True
def init():
from appbase.init import load_appbase from appbase.init import load_appbase
from rbac.init import load_rbac from rbac.init import load_rbac
load_appbase() load_appbase()