From 8aeb0723c1b91eb7c39b2753a78419931ec5f211 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 19 Apr 2026 21:57:31 +0200 Subject: [PATCH] Add location db adoption runbook --- docs/location-recorder.md | 70 +++++++++++++++++++-- docs/migration-notes.md | 7 +++ scripts/__init__.py | 1 + scripts/location_db_adopt.py | 118 +++++++++++++++++++++++++++++++++++ tests/test_location.py | 105 ++++++++++++++++++++++++++++++- 5 files changed, 293 insertions(+), 8 deletions(-) create mode 100644 scripts/__init__.py create mode 100644 scripts/location_db_adopt.py diff --git a/docs/location-recorder.md b/docs/location-recorder.md index ceff9bb..bb77ef2 100644 --- a/docs/location-recorder.md +++ b/docs/location-recorder.md @@ -1,6 +1,6 @@ # Location Recorder -本文档说明 `location recorder` 在 Python 项目中的当前数据库接管策略。 +本文档说明 `location recorder` 在 Python 项目中的当前数据库接管策略，以及 legacy SQLite 接管 runbook。 ## Legacy 事实基线 @@ -38,9 +38,33 @@ PRAGMA user_version = 2; - `20260419_01_location_baseline` +当前提供的最小脚本入口是： + +```bash +python scripts/location_db_adopt.py +``` + +如果你更喜欢模块方式运行，也可以用： + +```bash +python -m scripts.location_db_adopt +``` + +它只针对 `LOCATION_DATABASE_URL` 工作，并且遵守保守接管原则： + +- 本地已有 DB 文件：先校验，再接管 +- 本地没有 DB 文件：按新库初始化 +- 任一校验不通过：立即报错并停止 + ## 新数据库初始化 -对于一个全新 SQLite 数据库，执行： +如果本地不存在 `LOCATION_DATABASE_URL` 指向的 DB 文件： + +- 脚本会先创建父目录 +- 然后执行 Alembic `upgrade head` +- 最终建立 `location` 表与 `alembic_version` 表 + +手工执行时也等价于： ```bash alembic upgrade head @@ -52,9 +76,12 @@ alembic upgrade head 对于已经存在的 legacy SQLite 数据库： -1. 先确认其 `location` 表 schema 与 baseline 一致 -2. 旧库里的 `PRAGMA user_version = 2` 仅视为历史事实，不再继续沿用 -3. 确认无误后，对该数据库执行 `stamp`，而不是重新跑创建表 migration +1. 先确认 DB 文件存在 +2. 读取当前 DB 中 `location` 表的实际 schema +3. 与 baseline schema 做严格比对 +4. 再检查 `PRAGMA user_version` +5. 只有 schema 匹配且 `user_version = 2` 时，才执行 Alembic `stamp` +6. 接管完成后，后续 migration 才交给 Alembic 管理 示例： @@ -62,12 +89,37 @@ alembic upgrade head LOCATION_DATABASE_URL=sqlite:///./data/locationRecorder.db alembic stamp 20260419_01_location_baseline ``` +或直接执行脚本： + +```bash +LOCATION_DATABASE_URL=sqlite:///./data/locationRecorder.db python scripts/location_db_adopt.py +``` + 这样做的含义是： - 告诉 Alembic：这个数据库已经处于 baseline 结构 - 不修改已有 `location` 表数据 - 后续 migration 由 Alembic 接管 +## Fail Closed 原则 + +当前策略是保守接管，不做未知 legacy 状态的自动修复。 + +如果出现以下任一情况，脚本会直接报错并停止： + +- 找不到 `location` 表 +- `location` 表 schema 与 baseline 不一致 +- `PRAGMA user_version` 不等于 `2` +- 目标 DB 不是 SQLite URL + +当前不会尝试： + +- 自动修表 +- 自动调整 `user_version` +- 自动推断未知 legacy 状态 + +如果发生这些情况，应先人工确认数据库状态，再决定是否需要单独迁移或修复。 + ## 关于 `data/locationRecorder.db` 你本地放在 `data/locationRecorder.db` 的 legacy 样本库，可以用于： @@ -91,6 +143,12 @@ LOCATION_DATABASE_URL=sqlite:///./data/locationRecorder.db alembic stamp 2026041 - 构造一个“legacy 风格”的临时 SQLite 文件 - 建出同样的 `location` 表 - 设置 `PRAGMA user_version = 2` -- 再执行 Alembic `stamp` +- 再执行接管脚本中的 adopt 逻辑 + +同时也覆盖： + +- DB 文件不存在时的新库初始化路径 +- schema 不匹配时的失败路径 +- `user_version` 不匹配时的失败路径 这样可以验证接管路径，同时不污染真实样本库。 diff --git a/docs/migration-notes.md b/docs/migration-notes.md index a81fdcc..54079e4 100644 --- a/docs/migration-notes.md +++ b/docs/migration-notes.md @@ -61,6 +61,13 @@ CREATE TABLE location ( - [location-recorder.md](location-recorder.md) +当前还额外提供了一个最小 runbook / script 组合，用于保守接管 legacy location DB： + +- 先严格校验 schema +- 再严格校验 `PRAGMA user_version = 2` +- 只有全部匹配才执行 Alembic `stamp` +- 不匹配则直接失败，不自动修复 + ## 后续建议顺序 建议继续沿用既有迁移文档中的顺序： diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..02702eb --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Project helper scripts.""" diff --git a/scripts/location_db_adopt.py b/scripts/location_db_adopt.py new file mode 100644 index 0000000..bdedf3a --- /dev/null +++ b/scripts/location_db_adopt.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import sqlite3 +import sys +from pathlib import Path + +from alembic import command +from alembic.config import Config + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from app.config import get_settings + +LOCATION_BASELINE_REVISION = "20260419_01_location_baseline" +EXPECTED_USER_VERSION = 2 +EXPECTED_LOCATION_TABLE_INFO = [ + (0, "person", "TEXT", 1, None, 1), + (1, "datetime", "TEXT", 1, None, 2), + (2, "latitude", "REAL", 1, None, 0), + (3, "longitude", "REAL", 1, None, 0), + (4, "altitude", "REAL", 0, None, 0), +] + + +class LocationDatabaseAdoptionError(RuntimeError): + """Raised when a legacy location database does not match the expected baseline.""" + + +def _database_path_from_url(database_url: str) -> Path: + prefix = "sqlite:///" + if not database_url.startswith(prefix): + raise LocationDatabaseAdoptionError( + f"Only sqlite URLs are supported for location DB adoption, got: {database_url}" + ) + return Path(database_url[len(prefix) :]) + + +def _make_alembic_config(database_url: str) -> Config: + config = Config("alembic.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +def _location_table_exists(database_path: Path) -> bool: + conn = sqlite3.connect(database_path) + try: + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'location'" + ).fetchone() + return row is not None + finally: + conn.close() + + +def _fetch_location_table_info(database_path: Path) -> list[tuple]: + conn = sqlite3.connect(database_path) + try: + return list(conn.execute("PRAGMA table_info(location)")) + finally: + conn.close() + + +def _fetch_user_version(database_path: Path) -> int: + conn = sqlite3.connect(database_path) + try: + return conn.execute("PRAGMA user_version").fetchone()[0] + finally: + conn.close() + + +def validate_legacy_location_db(database_url: str) -> None: + database_path = _database_path_from_url(database_url) + if not database_path.exists(): + raise LocationDatabaseAdoptionError(f"Location DB file does not exist: {database_path}") + + if not _location_table_exists(database_path): + raise LocationDatabaseAdoptionError("Expected table 'location' was not found in the DB") + + table_info = _fetch_location_table_info(database_path) + if table_info != EXPECTED_LOCATION_TABLE_INFO: + raise LocationDatabaseAdoptionError( + "Location table schema does not match the expected baseline schema" + ) + + user_version = _fetch_user_version(database_path) + if user_version != EXPECTED_USER_VERSION: + raise LocationDatabaseAdoptionError( + f"Expected PRAGMA user_version = {EXPECTED_USER_VERSION}, got {user_version}" + ) + + +def adopt_or_initialize_location_db(database_url: str) -> str: + database_path = _database_path_from_url(database_url) + alembic_config = _make_alembic_config(database_url) + + if database_path.exists(): + validate_legacy_location_db(database_url) + command.stamp(alembic_config, LOCATION_BASELINE_REVISION) + return "adopted" + + database_path.parent.mkdir(parents=True, exist_ok=True) + command.upgrade(alembic_config, "head") + return "initialized" + + +def main() -> None: + settings = get_settings() + result = adopt_or_initialize_location_db(settings.location_database_url) + if result == "initialized": + print("Initialized a new location DB via Alembic upgrade head.") + else: + print("Validated legacy location DB and stamped Alembic baseline successfully.") + + +if __name__ == "__main__": + main() diff --git a/tests/test_location.py b/tests/test_location.py index 45a4241..c746254 100644 --- a/tests/test_location.py +++ b/tests/test_location.py @@ -10,8 +10,12 @@ from sqlalchemy.orm import sessionmaker import app.db from app.main import create_app - -LOCATION_BASELINE_REVISION = "20260419_01_location_baseline" +from scripts.location_db_adopt import ( + EXPECTED_USER_VERSION, + LOCATION_BASELINE_REVISION, + LocationDatabaseAdoptionError, + adopt_or_initialize_location_db, +) def _make_alembic_config(database_url: str) -> Config: @@ -170,3 +174,100 @@ def test_legacy_style_location_db_can_be_stamped_and_adopted( assert row_count == 1 engine.dispose() + + +def test_location_db_adoption_initializes_new_db(tmp_path: Path) -> None: + database_path = tmp_path / "new_location.db" + result = adopt_or_initialize_location_db(f"sqlite:///{database_path}") + + assert result == "initialized" + assert database_path.exists() + + conn = sqlite3.connect(database_path) + try: + revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] + location_table = conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'location'" + ).fetchone() + finally: + conn.close() + + assert revision == LOCATION_BASELINE_REVISION + assert location_table is not None + + +def test_location_db_adoption_validates_and_stamps_legacy_db(tmp_path: Path) -> None: + database_path = tmp_path / "legacy_location.db" + conn = sqlite3.connect(database_path) + conn.execute( + """ + CREATE TABLE location ( + person TEXT NOT NULL, + datetime TEXT NOT NULL, + latitude REAL NOT NULL, + longitude REAL NOT NULL, + altitude REAL, + PRIMARY KEY (person, datetime) + ) + """ + ) + conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") + conn.commit() + conn.close() + + result = adopt_or_initialize_location_db(f"sqlite:///{database_path}") + + assert result == "adopted" + + conn = sqlite3.connect(database_path) + try: + revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] + finally: + conn.close() + + assert revision == LOCATION_BASELINE_REVISION + + +def test_location_db_adoption_fails_closed_on_schema_mismatch(tmp_path: Path) -> None: + database_path = tmp_path / "bad_schema.db" + conn = sqlite3.connect(database_path) + conn.execute( + """ + CREATE TABLE location ( + person TEXT NOT NULL, + datetime TEXT NOT NULL, + latitude REAL NOT NULL, + longitude REAL NOT NULL, + PRIMARY KEY (person, datetime) + ) + """ + ) + conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") + conn.commit() + conn.close() + + with pytest.raises(LocationDatabaseAdoptionError, match="schema does not match"): + adopt_or_initialize_location_db(f"sqlite:///{database_path}") + + +def test_location_db_adoption_fails_closed_on_user_version_mismatch(tmp_path: Path) -> None: + database_path = tmp_path / "bad_user_version.db" + conn = sqlite3.connect(database_path) + conn.execute( + """ + CREATE TABLE location ( + person TEXT NOT NULL, + datetime TEXT NOT NULL, + latitude REAL NOT NULL, + longitude REAL NOT NULL, + altitude REAL, + PRIMARY KEY (person, datetime) + ) + """ + ) + conn.execute("PRAGMA user_version = 999") + conn.commit() + conn.close() + + with pytest.raises(LocationDatabaseAdoptionError, match="Expected PRAGMA user_version"): + adopt_or_initialize_location_db(f"sqlite:///{database_path}")