certflow.bootstrap 源代码
"""应用装配(Application Bootstrap / Composition Root)
阶段 B 的落地模块:把「数据库管理器从哪来、会话从哪来、Service 如何构造」
收口到单一的装配入口,消除散落在 ``BaseController`` / ``PrintQueueController``
里的隐式 ``DatabaseManager()`` 兜底与建表副作用。
设计约束(来自 ``tests/conftest.py`` 的书面约定):
- 提供工厂 ``build_context(db_manager=None) -> AppContext``。
- ``AppContext`` 至少暴露 ``.db_manager`` 与 ``.session`` 两个属性,
以兼容现有 ``app_context`` fixture 的调用方签名。
本模块刻意保持轻量:不做全局单例强制、不接管 Qt 对象、不 eager 创建有
副作用的 Service(如 ``PrinterManager`` 会起监控线程、
``CertificatePrintService`` 构造期 ``os.makedirs``)。这些仍以「按需 provider」
形式存在,未来逐 Controller 迁移时再逐一定义。
"""
from __future__ import annotations
from sqlalchemy.orm import Session
from certflow.utils.database import DatabaseManager
[文档]
class AppContext:
"""应用装配上下文
持有唯一的 ``DatabaseManager`` 与一个可选预创建的会话,作为
Controller / Service 依赖的统一来源。
Attributes:
db_manager: 数据库管理器实例(唯一权威来源)
session: 预创建的数据库会话(由 ``build_context`` 打开,
调用方可直接使用;不使用时由 ``close()`` 释放)
"""
def __init__(self, db_manager: DatabaseManager, session: Session | None = None) -> None:
self.db_manager = db_manager
self.session = session
[文档]
def get_session(self) -> Session:
"""获取一个数据库会话
若已持有会话(构造时预创建,或外部赋值 ``ctx.session = ...``)则复用之;
否则即时从 ``db_manager`` 新建。即时新建的会话由调用方负责 ``close()``。
Returns:
Session: 可用的 SQLAlchemy 会话
"""
if self.session is not None:
return self.session
return self.db_manager.get_session()
[文档]
def get_service(self, service_cls: type, *args: object, **kwargs: object) -> object:
"""以本上下文会话为首个参数构造一个 Service 实例
收口「``Service(self.session)`` 散落各处」的构造方式:调用方不再
直接触碰 ``self.session``,而是统一经此工厂获取 Service,使 session
来源(默认上下文 / 注入的 db_manager)成为唯一权威。
仅适用于构造签名首个位置参数为 ``Session`` 的 Service;首个参数非
会话的 Service(如 ``PrinterService(printer_name)``)不应走此工厂。
Args:
service_cls: Service 类(如 ``CertificateService``)
*args: 透传给 Service 构造的额外位置参数(排在 session 之后)
**kwargs: 透传给 Service 构造的关键字参数
Returns:
object: 构造好的 Service 实例
"""
return service_cls(self.get_session(), *args, **kwargs)
[文档]
def ensure_initialized(self) -> None:
"""确保数据库引擎与会话工厂已就绪
把「建表副作用」集中到此方法:仅在引擎尚未创建时才 ``init_db``,
避免 ``BaseController`` 每次构造都触发建表/迁移。
"""
if self.db_manager.engine is None:
self.db_manager.init_db(create_tables=True)
[文档]
def close(self) -> None:
"""释放本上下文持有的资源
关闭并清空当前持有的会话(无论预创建还是外部赋值),再关闭
``db_manager``。会话所有权统一由本上下文在 ``close`` 时回收。
"""
if self.session is not None:
self.session.close()
self.session = None
self.db_manager.close()
# 进程内默认上下文(惰性创建,集中收口隐式 DatabaseManager() 兜底)。
_default_context: AppContext | None = None
[文档]
def build_context(db_manager: DatabaseManager | None = None) -> AppContext:
"""构建应用装配上下文
约定入口:应用启动时由 ``main.init_database_manager()`` 注入已初始化的
``db_manager``;测试 fixture 注入临时库 ``db_manager``;二者皆无时惰性
创建一个默认 ``DatabaseManager()``(原本散落在 ``BaseController`` 的兜底
逻辑,现集中到此处单一来源)。
Args:
db_manager: 已配置(且按需 init_db)的数据库管理器;为 None 时惰性创建默认实例。
Returns:
AppContext: 装配好的应用上下文
"""
global _default_context
if db_manager is not None:
return AppContext(db_manager)
if _default_context is None:
_default_context = AppContext(DatabaseManager())
return _default_context
[文档]
def get_default_context() -> AppContext:
"""获取(惰性创建)进程内默认上下文
供 ``BaseController`` 等在没有显式/共享 db_manager 时统一取用,
替代原先各自 ``DatabaseManager()`` 的散落兜底。
Returns:
AppContext: 进程内默认上下文
"""
return build_context()
[文档]
def reset_default_context() -> None:
"""重置进程内默认上下文
供测试 teardown 调用,避免跨用例污染(原 ``_shared_db_manager`` 类变量
的同类问题)。
"""
global _default_context
_default_context = None