certflow.controllers.base_controller 源代码

"""控制器基类

提供所有业务控制器的基类实现，封装数据库会话管理、事务处理等通用功能.
"""

from __future__ import annotations

import types
from typing import Any

from loguru import logger
from sqlalchemy.orm import Session

from certflow.bootstrap import get_default_context
from certflow.utils.database import DatabaseManager




class BaseController:
    """控制器基类

    所有业务控制器的基类，提供统一的数据库会话管理和事务处理方法.
    支持上下文管理器协议，可自动管理会话生命周期.
    支持共享数据库管理器，实现全局WebDAV配置统一管理.

    Attributes:
        db_manager: 数据库管理器实例，负责创建和管理数据库连接
        session: SQLAlchemy数据库会话对象，用于执行数据库操作
        _owns_session: 是否拥有会话的所有权，决定close()时是否关闭会话

    Examples:
        >>> # 方式1: 使用上下文管理器(推荐)
        >>> with BaseController() as controller:
        ...     result = controller.session.query(User).all()
        ...     controller.commit()
        >>>
        >>> # 方式2: 手动管理会话
        >>> controller = BaseController()
        >>> try:
        ...     result = controller.session.query(User).all()
        ...     controller.commit()
        ... finally:
        ...     controller.close()
        >>>
        >>> # 方式3: 共享数据库管理器（支持WebDAV）
        >>> db_manager = DatabaseManager(webdav_config=config)
        >>> controller = BaseController(db_manager=db_manager)
        >>>
        >>> # 方式4: 使用全局共享管理器
        >>> BaseController.set_shared_db_manager(db_manager)
        >>> controller = BaseController()  # 自动使用共享管理器
    """

    # 类级别的共享数据库管理器（所有子类共享）
    _shared_db_manager: DatabaseManager | None = None



    @classmethod
    def set_shared_db_manager(cls, db_manager: DatabaseManager) -> None:
        """设置共享的数据库管理器（用于全局WebDAV配置）

        设置后，所有新创建的控制器如果没有显式传入db_manager，
        都会使用这个共享实例。

        Args:
            db_manager: 共享的数据库管理器实例

        Returns:
            None

        Examples:
            >>> # 在应用启动时设置
            >>> db_manager = DatabaseManager(webdav_config=config)
            >>> BaseController.set_shared_db_manager(db_manager)
        """
        cls._shared_db_manager = db_manager
        logger.info("已设置全局共享数据库管理器")




    @classmethod
    def get_shared_db_manager(cls) -> DatabaseManager | None:
        """获取共享的数据库管理器

        Returns:
            共享的数据库管理器实例，如果未设置则返回None
        """
        return cls._shared_db_manager




    @classmethod
    def clear_shared_db_manager(cls) -> None:
        """清除共享的数据库管理器

        将类级别的共享管理器置空，后续控制器将回退到独立管理器。

        Returns:
            None
        """
        cls._shared_db_manager = None
        logger.debug("已清除全局共享数据库管理器")


    def __init__(
        self,
        db_session: Session | None = None,
        db_manager: DatabaseManager | None = None,
        use_shared: bool = True,
    ):
        """初始化控制器

        支持多种使用方式：
        1. 自动创建会话：不传入任何参数，自动创建数据库会话
        2. 共享会话：传入外部会话，多个控制器共享同一个会话
        3. 共享管理器：传入db_manager或使用全局共享管理器
        4. 独立实例：不使用共享，创建独立数据库管理器

        Args:
            db_session: 外部传入的数据库会话对象，如果为None则自动创建新会话
                       传入外部会话时，控制器不会拥有会话所有权，
                       close()方法不会关闭该会话.
            db_manager: 外部传入的数据库管理器实例，如果为None则创建默认实例
                       传入db_manager可以共享WebDAV配置等资源
                       注意：如果同时传入db_session和db_manager，
                       db_session优先使用，db_manager仅用于引用
            use_shared: 是否使用类级别的共享管理器（默认True）
                       当db_manager为None且use_shared为True时，
                       会尝试使用共享管理器

        Examples:
            >>> # 自动创建（使用共享管理器）
            >>> controller = BaseController()
            >>>
            >>> # 使用独立管理器（不使用共享）
            >>> controller = BaseController(use_shared=False)
            >>>
            >>> # 传入外部管理器
            >>> controller = BaseController(db_manager=my_manager)
            >>>
            >>> # 传入外部会话
            >>> controller = BaseController(db_session=my_session)
        """
        # 1. 处理数据库管理器（优先级：参数 > 共享 > 默认上下文）
        if db_manager is not None:
            self.db_manager = db_manager
            logger.debug("使用显式传入的数据库管理器")
        elif use_shared and self._shared_db_manager is not None:
            self.db_manager = self._shared_db_manager
            logger.debug("使用全局共享数据库管理器")
        else:
            # 经阶段 B 的装配入口统一取默认上下文的 db_manager，
            # 消除原先散落的 DatabaseManager() 兜底（无 WebDAV 配置）。
            self.db_manager = get_default_context().db_manager
            logger.debug("使用默认应用上下文的数据库管理器")

        # 确保数据库已初始化（仅在引擎尚未创建时触发建表/迁移，
        # 避免每次构造都重复执行；应用引导阶段通常已完成）。
        self.db_manager.ensure_initialized_if_needed()

        # 2. 处理数据库会话（优先级：参数 > 自动创建）
        if db_session is not None:
            # 使用外部传入的会话
            self.session = db_session
            self._owns_session = False
            logger.debug("使用外部传入的数据库会话")
        else:
            # 自动创建会话
            self.session = self.db_manager.get_session()
            self._owns_session = True
            logger.debug("自动创建数据库会话")



    def __enter__(self) -> BaseController:
        """上下文管理器入口

        支持with语句，返回控制器实例自身.

        Returns:
            BaseController: 控制器实例自身

        Examples:
            >>> with BaseController() as controller:
            ...     # 使用controller进行操作
            ...     pass
        """
        return self




    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: types.TracebackType | None,
    ) -> None:
        """上下文管理器出口

        退出with代码块时自动关闭数据库会话.

        Args:
            exc_type: 异常类型，如果发生异常则为异常类
            exc_val: 异常实例，如果发生异常则为异常对象
            exc_tb: 异常追踪信息，如果发生异常则为traceback对象

        Returns:
            None
        """
        self.close()




    def close(self) -> None:
        """关闭数据库会话

        仅当控制器拥有会话所有权(即通过无参构造函数自动创建会话)时，
        才会关闭会话.外部传入的会话不会被关闭.

        Returns:
            None
        """
        if self._owns_session and self.session:
            self.session.close()
            logger.debug("数据库会话已关闭")




    def commit(self) -> None:
        """提交数据库事务

        提交当前会话中的所有更改.如果提交过程中发生异常，
        会自动回滚事务并重新抛出异常.

        Returns:
            None

        Raises:
            Exception: 事务提交失败时抛出原始异常
        """
        try:
            self.session.commit()
            logger.debug("事务提交成功")
        except Exception as e:
            self.session.rollback()
            logger.error(f"事务提交失败: {e}")
            raise




    def rollback(self) -> None:
        """回滚数据库事务

        撤销当前会话中所有未提交的更改.

        Returns:
            None
        """
        self.session.rollback()
        logger.debug("事务回滚")




    def get_session(self) -> Session:
        """获取当前会话

        Returns:
            Session: 当前使用的数据库会话
        """
        return self.session




    def get_service(self, service_cls: type, *args: object, **kwargs: object) -> object:
        """以当前控制器会话构造一个 Service 实例（收口 Service 构造来源）

        统一替代 ``XxxService(self.session)`` 的散落写法：session 来源经
        ``BaseController`` 收口，调用方不直接触碰 ``self.session``。

        仅适用于构造签名首个位置参数为 ``Session`` 的 Service；首个参数非
        会话的 Service（如 ``PrinterService(printer_name)``）不应走此工厂。

        Args:
            service_cls: Service 类（如 ``CertificateService``）
            *args: 透传给 Service 构造的额外位置参数（排在 session 之后）
            **kwargs: 透传给 Service 构造的关键字参数

        Returns:
            object: 构造好的 Service 实例
        """
        return service_cls(self.session, *args, **kwargs)


    # ============================================================
    # 静态门面：收口视图层对 services 工具的散落直连
    # （dict CSV 路径解析为各查表视图通用能力，下沉到基类控制器）
    # ============================================================



    @staticmethod
    def resolve_dict_csv_path(dict_key: str) -> Any:
        """解析字典 CSV 导出/导入路径（委托 ``dict_csv_sync.resolve_dict_csv_path``）。

        供 ``LookupTableManagerView`` 等查表视图替代对 ``certflow.services.
        dict_csv_sync`` 的直接依赖，统一经控制器门面取路径。
        """
        from certflow.services.dict_csv_sync import resolve_dict_csv_path

        return resolve_dict_csv_path(dict_key)




    def refresh_session(self) -> None:
        """刷新会话（如果会话已关闭，重新创建）

        用于在长时间运行的应用中恢复会话。

        Returns:
            None
        """
        if self.session is None or not hasattr(self.session, "is_active"):
            # 会话不存在或无效，重新创建
            self.session = self.db_manager.get_session()
            self._owns_session = True
            logger.debug("重新创建数据库会话")
        elif not self.session.is_active:
            # 会话不活跃，关闭旧会话并创建新会话
            self.close()
            self.session = self.db_manager.get_session()
            self._owns_session = True
            logger.debug("刷新数据库会话")




    def is_session_active(self) -> bool:
        """检查会话是否活跃

        Returns:
            bool: 会话是否存在且活跃
        """
        return (
            self.session is not None
            and hasattr(self.session, "is_active")
            and self.session.is_active
        )