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