certflow.utils.database module

数据库管理工具模块

提供数据库引擎创建、会话管理和表初始化等功能， 基于SQLAlchemy的SQLite数据库实现。 支持从WebDAV服务器同步数据库文件。

class certflow.utils.database.NutstoreWebDAVConfig(url, username, password, remote_path='CertFlow/backups', keep_backups=10)

基类：object

坚果云WebDAV配置类

参数:

• url (str)

• username (str)

• password (str)

• remote_path (str)

• keep_backups (int)

get_auth()

获取HTTP Basic认证

返回:

用于 WebDAV 请求的 Basic Auth 认证对象。

返回类型:

HTTPBasicAuth

get_remote_url(filename='')

获取完整的远程URL

参数:

filename (str) -- 文件名，为空时返回目录 URL。

返回:

完整的远程 URL 路径。

返回类型:

str

class certflow.utils.database.DatabaseManager(db_path=None, webdav_config=None, auto_backup_before_sync=True)

基类：object

数据库管理器

管理数据库引擎、会话工厂和表创建，提供统一的数据库操作入口。 支持数据库初始化、表迁移(自动补齐缺失列)和会话管理。 支持从WebDAV服务器同步数据库文件。

参数:

• db_path (str | None)

• webdav_config (NutstoreWebDAVConfig | None)

• auto_backup_before_sync (bool)

db_path

数据库文件路径

engine

SQLAlchemy引擎实例，负责数据库连接

SessionLocal

会话工厂实例，用于创建数据库会话

webdav_config

WebDAV配置（如果启用）

示例

>>> from certflow.utils.database import DatabaseManager
>>>
>>> # 创建数据库管理器
>>> db = DatabaseManager("path/to/database.db")
>>>
>>> # 初始化数据库
>>> db.init_db(create_tables=True)
>>>
>>> # 从WebDAV拉取数据库
>>> webdav_config = NutstoreWebDAVConfig(
...     url="https://dav.jianguoyun.com/dav/",
...     username="user@example.com",
...     password="app_password"
... )
>>> db.init_db(sync_from_webdav=True)
>>>
>>> # 获取会话并执行操作
>>> session = db.get_session()
>>> try:
...     result = session.query(SalePlan).all()
... finally:
...     session.close()

init_db(create_tables=True, sync_from_webdav=False, force_sync=False)

初始化数据库

创建数据库目录、SQLAlchemy引擎和会话工厂， 可选创建所有数据表。已有表会自动补齐缺失列。 可选从WebDAV同步数据库文件。

参数:

• create_tables (bool) -- 是否创建数据表，默认为True

• sync_from_webdav (bool) -- 是否从WebDAV同步数据库，默认为False

• force_sync (bool) -- 是否强制同步（即使本地数据库存在也尝试从远程拉取）

抛出:

Exception -- 当数据库初始化失败时抛出异常

返回类型:

None

示例

>>> db = DatabaseManager("test.db")
>>> db.init_db(create_tables=True)
>>> print("数据库初始化完成")

ensure_initialized_if_needed()

仅在引擎尚未创建时初始化数据库（幂等、低开销）

供装配层（BaseController / AppContext）在「不确定当前是否已完成 引导」的构造路径上调用：已初始化则直接跳过，未初始化则补建表/迁移。 区别于 init_db 的「总是执行完整初始化」，本方法避免在每次 Controller 构造时重复跑建表与迁移流程。

返回类型:

None

warn_if_base_data_empty()

检查基础字典底座是否为空，为空则给出 seed 提示（只读、不自动播种）。

对应 DATABASE_BASE_DATA.md §4.2 的「可选加固」：在 init_db 之后挂载 一个轻量检查——若 material_grades / model_param_mappings / caliber_mappings 任一为空，记录 WARNING 并提示运行 scripts/data/seed_base_data.py。

刻意**不**自动播种（§4.2 明确「不要每次启动自动 seed」，避免依赖 xlsx / 拖慢启动）；自动播种由 main._maybe_seed_base_data 在 GUI 启动路径单独负责。 CLI / 双库等不走自动播种的入口调用本方法即可获得「空库提示」安全网。

返回:

字典表是否为空（True=空，需 seed）。

返回类型:

bool

static commit_with_retry(session, max_attempts=12)

带退避重试的 session.commit，专治瞬时写锁。

直接在业务 session 上调用，遇 database is locked 时指数退避 重试（base 0.05s / 上限 0.8s / 累计约 8s）。配合 WAL+busy_timeout 使用，几乎可消除打印流程中的写保护锁现象。

参数:

• session (Any) -- SQLAlchemy Session（其引擎需已设 busy_timeout）。

• max_attempts (int) -- 最大尝试次数。

抛出:

非锁类异常或重试耗尽后仍失败时原样抛出。 --

返回类型:

None

static flush_with_retry(session, max_attempts=12)

带退避重试的 session.flush（写库前先落盘也可能需要抢锁）。

参数:

• session (Any)

• max_attempts (int)

返回类型:

None

pull_from_webdav(backup_before_pull=True)

从坚果云拉取最新数据库备份

参数:

backup_before_pull (bool) -- 拉取前是否备份本地数据库

返回:

是否成功拉取并替换

返回类型:

bool

push_to_webdav(backup_remote=True)

上传本地数据库到坚果云

参数:

backup_remote (bool) -- 是否备份远程数据库（保留旧版本）

返回:

是否成功上传

返回类型:

bool

sync_with_webdav(direction='pull')

与WebDAV服务器同步数据库

参数:

direction (str) -- 同步方向，"pull"（从服务器拉取）或 "push"（推送到服务器）

返回:

是否同步成功

返回类型:

bool

check_remote_status()

检查远程备份状态

返回:

包含远程备份信息的字典

返回类型:

dict[str, Any]

get_session()

获取数据库会话

如果会话工厂未初始化则先初始化数据库，然后创建并返回新会话。

返回:

SQLAlchemy数据库会话实例，可用于执行数据库操作

返回类型:

Session

示例

>>> db = DatabaseManager()
>>> session = db.get_session()
>>> try:
...     # 执行数据库操作
...     result = session.query(SalePlan).filter_by(id=1).first()
... finally:
...     session.close()

close()

关闭数据库连接

释放数据库引擎资源，关闭所有活动连接。

示例

>>> db = DatabaseManager()
>>> db.init_db()
>>> # 使用数据库...
>>> db.close()  # 程序退出前释放资源

返回类型:

None