diff --git a/app/db.py b/app/db.py index 8ca241b..6faf9b2 100644 --- a/app/db.py +++ b/app/db.py @@ -1,7 +1,7 @@ from collections.abc import Generator from pathlib import Path -from sqlalchemy import create_engine, event +from sqlalchemy import Engine, create_engine, event from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker from app.config import get_settings @@ -34,10 +34,41 @@ engine = create_sqlite_engine(get_settings().database_url) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) +def check_database_integrity(target_engine: Engine = engine) -> str: + raw_connection = target_engine.raw_connection() + try: + cursor = raw_connection.driver_connection.cursor() + try: + cursor.execute("PRAGMA integrity_check") + result = cursor.fetchone() + finally: + cursor.close() + finally: + raw_connection.close() + + status = result[0] if result else "missing_result" + if status != "ok": + raise RuntimeError(f"SQLite integrity check failed: {status}") + return status + + +def checkpoint_sqlite_wal(target_engine: Engine = engine) -> bool: + if str(target_engine.url) == "sqlite:///:memory:": + return False + if target_engine.url.get_backend_name() != "sqlite": + return False + + with target_engine.begin() as connection: + connection.exec_driver_sql("PRAGMA wal_checkpoint(TRUNCATE)") + return True + + def init_db() -> None: import app.models # noqa: F401 Base.metadata.create_all(engine) + check_database_integrity(engine) + checkpoint_sqlite_wal(engine) def get_db_session() -> Generator[Session]: diff --git a/docs/MVP-WorkOrders.md b/docs/MVP-WorkOrders.md index 8162244..3256c36 100644 --- a/docs/MVP-WorkOrders.md +++ b/docs/MVP-WorkOrders.md @@ -532,6 +532,28 @@ fix: 完善失败重试与跳过边界 fix: 增强 SQLite 数据库恢复与异常提示 ``` +完成记录: + +```text +完成日期:2026-07-03 +相关 commit:fix: 增强 SQLite 数据库恢复与异常提示 +验证命令: +- .venv/bin/python -m pytest tests/unit/test_db_stability.py tests/integration/test_task_recovery.py -q +- docker compose up -d --build +- curl -f http://localhost:8000/health +- docker compose exec -T app python 检查 PRAGMA integrity_check 与 wal_checkpoint(TRUNCATE) +- .venv/bin/python -m pytest tests/unit tests/integration -q +- .venv/bin/python -m pytest tests/unit tests/integration --cov=app --cov-branch --cov-report=term-missing +验收结论: +- 启动时执行 PRAGMA integrity_check,异常时抛出明确 RuntimeError,不新增配置开关。 +- 启动建表后执行 PRAGMA wal_checkpoint(TRUNCATE),将 WAL 写入主库并截断 WAL 文件。 +- Docker 重建后,独立脚本可读到 WO-04 默认规模任务 4bdf36df-8ae4-4059-af98-602f4872dbaa 与 378b173a-a70f-4da1-b235-03e7632e614a。 +- Uvicorn 进程不再持有 deleted app.db-wal/app.db-shm 文件句柄。 +- Docker 重启后 running 任务恢复逻辑仍通过测试。 +遗留问题: +- 数据库损坏备份策略和本地重置说明需要在 WO-09 用户操作文档中补充。 +``` + ### WO-07 导出链路验收与文件质量优化 优先级:P1 diff --git a/tests/unit/test_db_stability.py b/tests/unit/test_db_stability.py new file mode 100644 index 0000000..8003fc3 --- /dev/null +++ b/tests/unit/test_db_stability.py @@ -0,0 +1,65 @@ +import pytest + +from app.db import check_database_integrity, checkpoint_sqlite_wal, create_sqlite_engine + + +def test_check_database_integrity_returns_ok_for_valid_sqlite_database(): + engine = create_sqlite_engine("sqlite:///:memory:") + try: + assert check_database_integrity(engine) == "ok" + finally: + engine.dispose() + + +def test_check_database_integrity_raises_when_sqlite_reports_problem(monkeypatch): + class FakeCursor: + def execute(self, _sql): + return None + + def fetchone(self): + return ("database disk image is malformed",) + + def close(self): + return None + + class FakeConnection: + def cursor(self): + return FakeCursor() + + class FakeRawConnection: + driver_connection = FakeConnection() + + def close(self): + return None + + class FakeEngine: + def raw_connection(self): + return FakeRawConnection() + + with pytest.raises(RuntimeError, match="SQLite integrity check failed"): + check_database_integrity(FakeEngine()) + + +def test_checkpoint_sqlite_wal_truncates_wal_for_file_database(tmp_path): + db_path = tmp_path / "app.db" + engine = create_sqlite_engine(f"sqlite:///{db_path}") + try: + with engine.begin() as connection: + connection.exec_driver_sql("CREATE TABLE demo (id INTEGER PRIMARY KEY, name TEXT)") + connection.exec_driver_sql("INSERT INTO demo (name) VALUES ('ok')") + + assert checkpoint_sqlite_wal(engine) is True + + with engine.connect() as connection: + rows = connection.exec_driver_sql("SELECT name FROM demo").all() + assert rows == [("ok",)] + finally: + engine.dispose() + + +def test_checkpoint_sqlite_wal_skips_in_memory_database(): + engine = create_sqlite_engine("sqlite:///:memory:") + try: + assert checkpoint_sqlite_wal(engine) is False + finally: + engine.dispose()