feat: main 集成(路由挂载/trace 中间件/lifespan) + B7 挂账收口(引擎工厂/统一错误体/锁公共化/L3 缓存 DEL/死代码清理/处置原子事务/启动期 debug 校验, B7)
This commit is contained in:
@@ -0,0 +1,33 @@
|
||||
"""SQLAlchemy 引擎工厂(B7 · 开发计划挂账⑥)。
|
||||
|
||||
按库名缓存单例 Engine:deps / risk / simulate 路由每请求实例化 Repository 时
|
||||
复用同一连接池,不再每次 create_engine(B6 复审 P3:实例化点泄漏);应用
|
||||
shutdown 经 dispose_engines 统一释放。测试直传 engine= 参数的用法不受影响;
|
||||
monkeypatch settings 后须先 dispose_engines() 清缓存。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import lru_cache
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.engine import Engine
|
||||
|
||||
from app.config.settings import settings
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def get_engine(database: str) -> Engine:
|
||||
"""取指定库的单例 Engine(进程内共享连接池)。"""
|
||||
pwd = settings.mysql_password
|
||||
auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user
|
||||
url = (
|
||||
f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}"
|
||||
f"/{database}?charset=utf8mb4"
|
||||
)
|
||||
return create_engine(url, pool_pre_ping=True)
|
||||
|
||||
|
||||
def dispose_engines() -> None:
|
||||
"""释放全部缓存 Engine(lifespan shutdown / 测试隔离)。"""
|
||||
get_engine.cache_clear()
|
||||
Reference in New Issue
Block a user