certflow.cli.base_command 源代码

# src/certflow/cli/base_command.py
"""CLI命令基类 - 提供统一的数据库管理器初始化"""

from __future__ import annotations

import os

from dotenv import load_dotenv
from loguru import logger
from sqlalchemy.orm import Session

from certflow.utils.database import DatabaseManager, NutstoreWebDAVConfig




def get_cli_db_manager(
    enable_webdav: bool = True, create_tables: bool = True, sync_from_webdav: bool = True
) -> DatabaseManager:
    """获取 CLI 命令使用的数据库管理器

    Args:
        enable_webdav: 是否启用 WebDAV 同步（默认 True）
        create_tables: 是否创建数据表（默认 True）
        sync_from_webdav: 是否从 WebDAV 同步（默认 True）

    Returns:
        初始化好的数据库管理器实例
    """
    load_dotenv()

    webdav_config = None
    if enable_webdav:
        url = os.getenv("NUTSTORE_WEBDAV_URL")
        username = os.getenv("NUTSTORE_WEBDAV_USER")
        password = os.getenv("NUTSTORE_WEBDAV_PASSWORD")

        if all([url, username, password]):
            logger.info("检测到 WebDAV 配置，CLI 将启用远程同步")
            webdav_config = NutstoreWebDAVConfig(
                url=url,
                username=username,
                password=password,
                remote_path=os.getenv("NUTSTORE_REMOTE_PATH", "CertFlow/backups"),
                keep_backups=int(os.getenv("NUTSTORE_KEEP_BACKUPS", "10")),
            )

    db_manager = DatabaseManager(webdav_config=webdav_config, auto_backup_before_sync=True)

    db_manager.init_db(
        create_tables=create_tables,
        sync_from_webdav=sync_from_webdav and bool(webdav_config),
        force_sync=False,
    )

    # CLI 路径不自动播种：空库时给出 seed 提示安全网（只读，不自动 seed）
    db_manager.warn_if_base_data_empty()

    return db_manager





def get_cli_session(enable_webdav: bool = True) -> Session:
    """获取数据库会话（快捷方法）

    Args:
        enable_webdav: 是否启用 WebDAV 同步（默认 True）。

    Returns:
        Session: SQLAlchemy 数据库会话对象。
    """
    return get_cli_db_manager(enable_webdav).get_session()