appdb/migrate.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Move the application database between backends -- admin panel phase 5.
``docs/admin-panel-architecture.md`` §5.4 is the design contract this
module implements: ``SQLite -> SQL Server``, ``SQLite -> PostgreSQL``, and
the reverse, as one shot, offline, verified, and reversible by virtue of
the source being left untouched. ``scripts/migrate_app_db.py`` is the CLI
wrapper; every real decision lives here so it can be tested directly,
against real SQLAlchemy engines on both sides, with no mock at the
boundary under test.
Type mapping (§4 -- decided here, next to the schema)
-------------------------------------------------------
Every column :mod:`appdb.models` declares is ``String``, ``Integer``, or
``Text`` -- see that module's docstring. None of the three ambiguous
SQLAlchemy types the architecture doc calls out by name (``Boolean``,
``DateTime``, a bespoke JSON column type) appears anywhere in this schema:
timestamps are ISO-8601 strings in ``String`` columns (``appdb.key_store``
etc. all write ``datetime.now(timezone.utc).isoformat()``), JSON payloads
are pre-serialised to text before they are ever stored
(``denied_columns_json``, ``content_json``), and what would elsewhere be a
boolean flag is instead a nullable timestamp tombstone (``disabled_at``,
``revoked_at``) -- a deliberate choice in :mod:`appdb.models` itself that
sidesteps the "``1`` on one side, ``true`` on the other" hazard rather than
needing this module to paper over it. Copying therefore never touches raw
row values with anything but plain ``str``/``int``/``None`` in transit:
every read goes through :func:`sqlalchemy.select` against the *same*
``Table`` objects :mod:`appdb.models` defines, and every write goes
through the same tables' ``.insert()``, so SQLAlchemy Core's own,
already-correct per-dialect type compilation is what binds each value on
the way in and out -- never a hand-rolled string-to-native conversion this
module would have to get right a second time. If a future phase adds a
genuine ``Boolean``/``DateTime``/JSON-typed column, the mapping decision
belongs there, in :mod:`appdb.models`, next to that column -- not here.
Identifiers are preserved, never renumbered (§3)
---------------------------------------------------
Every row is copied with its primary key given explicitly
(``table.insert()`` with the source's own column values, including the
key column) rather than left for the target to assign -- the only way a
feedback row's ``config_version_id`` or a config version's own
``version_id`` can still resolve on the far side. For the two
autoincrement integer primary keys this schema has
(:data:`AUTOINCREMENT_PK_COLUMNS`), :func:`_reset_autoincrement` reseeds
the target's own sequence to ``MAX(id) + 1`` afterwards, per backend, so
the *next* row the running application inserts there (with no explicit id
of its own) never collides with one this tool just copied in.
It must be verifiable (§5)
------------------------------
:func:`verify_migration` is independent of the copy step -- it re-reads
both databases from scratch and compares, per table, row counts and a
content hash. The hash reuses :func:`eval.fingerprint.fingerprint_dataframe`
rather than a second hasher: the "same data regardless of row/column
order, robust to the usual numpy/Decimal/None noise" property that module
was built for is exactly what a cross-backend row comparison needs too.
:func:`hash_database` gives the same idea one level up -- a single
fingerprint of the *entire* source database, taken both before and after
every run (even a failed one) to prove this tool never mutated the source
it read from (§6).
Maintenance mode does not exist yet (§7)
--------------------------------------------
Phase 6 defines it. Until then, :func:`check_quiescent` (run as part of
:func:`export_database`, before anything is written anywhere) refuses
whenever the source's own timestamp columns show activity newer than
``cfg.settings.migration_quiet_window_seconds`` -- a refusal that names
maintenance mode explicitly in its message, rather than a tool that
silently copies past writes still in flight.
Schema versions must match (§8)
------------------------------------
:func:`current_schema_version` is Alembic's own head revision for
``appdb/migrations`` (not a number invented for this phase) -- read from
the migration scripts on disk, independent of what either database's own
``alembic_version`` table happens to say. Every export carries the
exporting process's current head as its stamp; :func:`check_schema_version`
refuses to import an export whose stamp does not match the *importing*
process's own head, exactly the case where the two sides are running code
that disagree about the shape of this schema.
"""
from __future__ import annotations
import hashlib
import json
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import pandas as pd
from sqlalchemy import Table, func, inspect, select, text
from sqlalchemy.engine import Engine
import config as cfg
from appdb.engine import _canonical_endpoint, build_engine, redact_url
from appdb.models import (
access_requests,
admin_api_keys,
admin_principal_roles,
config_bundle_versions,
metadata,
turn_confirmations,
turn_feedback,
)
from eval.fingerprint import fingerprint_dataframe
#: Every table this tool moves, in the order rows are written on import.
#: None of these columns are declared with a real ``ForeignKey`` constraint
#: (see ``appdb/models.py`` -- the relationships are enforced by the
#: application, not the database), so this order is not required for
#: referential integrity to hold at the database level, but it is kept
#: parent-before-child anyway (a config version before the feedback rows
#: that may cite its id) so a partial copy interrupted mid-run is at least
#: never backwards.
TABLES_IN_MIGRATION_ORDER: tuple[Table, ...] = (
admin_api_keys,
admin_principal_roles,
config_bundle_versions,
turn_feedback,
access_requests,
turn_confirmations,
)
#: ``{table_name: primary_key_column_name}`` for every table whose primary
#: key is a database-assigned autoincrement integer -- the only columns
#: :func:`_reset_autoincrement` ever needs to touch. ``admin_api_keys``
#: (keyed by ``key_sha256``) and ``admin_principal_roles`` (keyed by
#: ``(principal_id, capability)``) both have caller-supplied, non-integer
#: keys and need no reseeding.
AUTOINCREMENT_PK_COLUMNS: dict[str, str] = {
"config_bundle_versions": "version_id",
"turn_feedback": "feedback_id",
"access_requests": "request_id",
"turn_confirmations": "confirmation_id",
}
#: Column-name suffix :func:`_max_activity_timestamp` treats as a
#: timestamp worth scanning for recent activity. Every such column in this
#: schema stores ``datetime.now(timezone.utc).isoformat()`` (see the
#: module docstring's "Type mapping" section).
_TIMESTAMP_COLUMN_SUFFIX = "_at"
class MigrationRefusedError(RuntimeError):
"""The migration tool refused to proceed. The message says why --
recent write activity, a schema-version mismatch, a non-empty target,
or source and target naming the same database."""
@dataclass(frozen=True)
class TableSnapshot:
"""One table's rows, read at one point in time, plus its content hash."""
name: str
columns: tuple[str, ...]
rows: tuple[dict[str, Any], ...]
row_count: int
content_hash: str
@dataclass(frozen=True)
class MigrationExport:
"""Everything read from a source database -- "the exported artefact"
(§9). Contains every issued key's ``key_sha256`` and every key's
``denied_columns_json`` ACL verbatim: not a secret in the ``.env``
sense, but not something to leave in a shared folder either -- see
``scripts/migrate_app_db.py`` and ``docs/deployment-runbook.md``.
"""
schema_version: str
exported_at: str
tables: dict[str, TableSnapshot]
@dataclass(frozen=True)
class TableVerification:
"""One table's post-copy comparison -- row counts and content hashes,
read independently from both databases."""
table: str
source_row_count: int
target_row_count: int
source_hash: str
target_hash: str
@property
def ok(self) -> bool:
return (
self.source_row_count == self.target_row_count
and self.source_hash == self.target_hash
)
@dataclass(frozen=True)
class VerificationReport:
"""Every table's :class:`TableVerification` from one comparison pass."""
tables: tuple[TableVerification, ...]
@property
def ok(self) -> bool:
return all(t.ok for t in self.tables)
@dataclass(frozen=True)
class MigrationResult:
"""The outcome of one :func:`run_migration` call -- always returned,
never raised, for every refusal this module recognises (§11's "must say
so, loudly, rather than reporting success" applies to a caller checking
``.ok``, not to catching an exception). A genuinely unexpected failure
(e.g. the target is unreachable) still propagates as whatever
SQLAlchemy raises -- that is an infrastructure fault, not a decision
this tool makes."""
dry_run: bool
ok: bool
message: str
schema_version: str
source_hash_before: str
source_hash_after: str
export_row_counts: dict[str, int]
verification: VerificationReport | None
# ---------------------------------------------------------------------------
# Schema version (§8)
# ---------------------------------------------------------------------------
def current_schema_version() -> str:
"""Alembic's current head revision for ``appdb/migrations`` -- read from
the migration scripts on disk (never from either database's own
``alembic_version`` table), so this is exactly the schema version the
*running code* expects, independent of what has or has not actually
been applied to any particular database.
"""
from alembic.config import Config
from alembic.script import ScriptDirectory
repo_root = Path(__file__).resolve().parent.parent
alembic_cfg = Config()
alembic_cfg.set_main_option(
"script_location", str(repo_root / "appdb" / "migrations")
)
return ScriptDirectory.from_config(alembic_cfg).get_current_head()
def check_schema_version(export: MigrationExport) -> None:
"""Refuse if *export*'s stamp does not match this installation's own
current schema version (§8).
Raises
------
MigrationRefusedError
"""
current = current_schema_version()
if export.schema_version != current:
raise MigrationRefusedError(
"schema-version mismatch: this export was produced at schema "
f"revision {export.schema_version!r}, but this installation's "
f"migrations are at {current!r}. Run this installation's "
"migrations against the target (`alembic upgrade head`) -- or "
"re-export from a build whose revision matches -- before "
"importing. Silently loading a different shape is how a "
"subtly broken installation gets created."
)
# ---------------------------------------------------------------------------
# Reading a table (used for both export and verification)
# ---------------------------------------------------------------------------
def _table_columns(table: Table) -> tuple[str, ...]:
return tuple(c.name for c in table.columns)
def _rows_as_dataframe(rows: list[dict[str, Any]], columns: tuple[str, ...]) -> pd.DataFrame:
"""A DataFrame with exactly *columns*, even when *rows* is empty --
:func:`eval.fingerprint.fingerprint_dataframe` gives an empty frame a
well-defined fingerprint only when its column set is present."""
if rows:
return pd.DataFrame(rows, columns=list(columns))
return pd.DataFrame({c: pd.Series([], dtype="object") for c in columns})
def _empty_snapshot(table: Table) -> TableSnapshot:
columns = _table_columns(table)
return TableSnapshot(
name=table.name,
columns=columns,
rows=(),
row_count=0,
content_hash=fingerprint_dataframe(_rows_as_dataframe([], columns)),
)
def _table_snapshot(conn, table: Table) -> TableSnapshot:
columns = _table_columns(table)
rows = [dict(r) for r in conn.execute(select(table)).mappings().all()]
df = _rows_as_dataframe(rows, columns)
return TableSnapshot(
name=table.name,
columns=columns,
rows=tuple(rows),
row_count=len(rows),
content_hash=fingerprint_dataframe(df),
)
def _snapshot_all_tables(engine: Engine) -> dict[str, TableSnapshot]:
insp = inspect(engine)
out: dict[str, TableSnapshot] = {}
with engine.connect() as conn:
for table in TABLES_IN_MIGRATION_ORDER:
out[table.name] = (
_table_snapshot(conn, table)
if insp.has_table(table.name)
else _empty_snapshot(table)
)
return out
# ---------------------------------------------------------------------------
# Non-mutation proof (§6) -- a whole-database fingerprint
# ---------------------------------------------------------------------------
def hash_database(url: str) -> str:
"""A single fingerprint of every migrated table in the database at
*url* -- computed before and after every :func:`run_migration` call
(success or failure) to prove the source was never mutated.
"""
engine = build_engine(url)
try:
per_table = {
name: snapshot.content_hash
for name, snapshot in _snapshot_all_tables(engine).items()
}
payload = json.dumps(per_table, sort_keys=True)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
finally:
engine.dispose()
# ---------------------------------------------------------------------------
# Recent-write-activity refusal (§7) -- see check_quiescent on why this
# is not replaced by maintenance mode
# ---------------------------------------------------------------------------
def _max_activity_timestamp(export: MigrationExport) -> datetime | None:
"""The most recent parseable ``*_at`` timestamp across every row of
every table in *export*, or ``None`` if there is none (an empty
database, or one whose rows carry no timestamp at all)."""
latest: datetime | None = None
for snapshot in export.tables.values():
timestamp_columns = [c for c in snapshot.columns if c.endswith(_TIMESTAMP_COLUMN_SUFFIX)]
if not timestamp_columns:
continue
for row in snapshot.rows:
for col in timestamp_columns:
value = row.get(col)
if not value:
continue
try:
parsed = datetime.fromisoformat(value)
except (TypeError, ValueError):
continue
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
if latest is None or parsed > latest:
latest = parsed
return latest
def check_quiescent(export: MigrationExport) -> None:
"""Refuse if *export* shows write activity inside the quiet window
(``cfg.settings.migration_quiet_window_seconds``).
Raises
------
MigrationRefusedError
"""
latest = _max_activity_timestamp(export)
if latest is None:
return
window = cfg.settings.migration_quiet_window_seconds
age_seconds = (datetime.now(timezone.utc) - latest).total_seconds()
if age_seconds < window:
raise MigrationRefusedError(
"the source application database has write activity from "
f"{max(age_seconds, 0.0):.1f}s ago, inside the {window:.0f}s "
"quiet window this tool requires -- refusing to migrate. This "
"tool requires the application to be stopped or in maintenance "
"mode: a write that lands in the source after this tool has "
"read past it is lost with no error at all. Maintenance mode "
"exists (api/maintenance.py) but its flag lives in the "
"server's own process memory, so this tool -- a separate "
"process -- cannot observe it; turning maintenance on does "
"not lift this refusal. Stop the application (or wait for it "
"to go quiet) and re-run."
)
# ---------------------------------------------------------------------------
# Export (read-only against the source)
# ---------------------------------------------------------------------------
def export_database(source_url: str) -> MigrationExport:
"""Read every row of every migrated table from *source_url*.
Read-only: nothing here ever writes to the source. Refuses
(:class:`MigrationRefusedError`) if the source shows write activity
inside the quiet window (:func:`check_quiescent`) -- checked here,
before any target is ever touched.
"""
engine = build_engine(source_url)
try:
tables = _snapshot_all_tables(engine)
finally:
engine.dispose()
export = MigrationExport(
schema_version=current_schema_version(),
exported_at=datetime.now(timezone.utc).isoformat(),
tables=tables,
)
check_quiescent(export)
return export
def export_to_json(export: MigrationExport) -> str:
"""Serialise *export* -- the on-disk shape of "the exported artefact"
(§9) that ``scripts/migrate_app_db.py`` writes to a temporary file.
Sensitive: see the module docstring's :class:`MigrationExport` note."""
payload = {
"schema_version": export.schema_version,
"exported_at": export.exported_at,
"tables": {
name: {
"columns": list(snapshot.columns),
"rows": [dict(r) for r in snapshot.rows],
"row_count": snapshot.row_count,
"content_hash": snapshot.content_hash,
}
for name, snapshot in export.tables.items()
},
}
return json.dumps(payload, indent=2, sort_keys=True)
def export_from_json(payload: str) -> MigrationExport:
"""The inverse of :func:`export_to_json` -- rebuilds a
:class:`MigrationExport` from a previously written export file."""
data = json.loads(payload)
tables = {
name: TableSnapshot(
name=name,
columns=tuple(t["columns"]),
rows=tuple(t["rows"]),
row_count=t["row_count"],
content_hash=t["content_hash"],
)
for name, t in data["tables"].items()
}
return MigrationExport(
schema_version=data["schema_version"],
exported_at=data["exported_at"],
tables=tables,
)
# ---------------------------------------------------------------------------
# Target-side checks
# ---------------------------------------------------------------------------
def _redacted(url: str) -> str:
"""*url* with its password replaced by ``***``, for a message a human
will read.
A refusal has to name the two URLs -- an operator staring at "these
resolve to the same database" needs to see *which* two -- but the
complete URL of a networked backend carries the credential in it
(``postgresql://user:pw@host/db``). The identity and the secret are
not the same thing, and only one of them belongs in a message this
module hands back to a caller: :func:`run_migration` copies it
verbatim into :attr:`MigrationResult.message`, and ``scripts/
migrate_app_db.py`` both prints and logs that.
Uses SQLAlchemy's own :meth:`~sqlalchemy.engine.URL.render_as_string`
rather than a regular expression over the text, for the same reason
``appdb/engine.py`` compares URLs after parsing rather than as
strings: the parser already knows where the password is in every URL
shape this project can be pointed at, and a regex would be a second,
worse implementation of that knowledge. ``scripts/
verify_deployment.py`` already redacts the warehouse URL this way.
A URL SQLAlchemy cannot parse falls back to a bare placeholder rather
than to the raw string -- an unparseable value is exactly the case
where a hand-rolled redaction would be most likely to miss something,
and the identity it would have carried is not trustworthy anyway.
"""
return redact_url(url)
def check_not_same_database(source_url: str, target_url: str) -> None:
"""Refuse if *source_url* and *target_url* resolve to the same
database -- reusing :func:`appdb.engine._canonical_endpoint`'s own
after-parsing comparison (``localhost``/``127.0.0.1``, a resolved
SQLite path, ...) rather than a second, string-only implementation.
The refusal names both URLs with their passwords redacted -- see
:func:`_redacted` for why a message built to be displayed must not
carry the credential that happens to live in the same string.
Raises
------
MigrationRefusedError
"""
if _canonical_endpoint(source_url) == _canonical_endpoint(target_url):
raise MigrationRefusedError(
"source and target resolve to the same database -- nothing to "
f"migrate. source: {_redacted(source_url)!r}; "
f"target: {_redacted(target_url)!r}."
)
def check_target_is_empty(engine: Engine) -> None:
"""Refuse if any migrated table already has rows in it -- this tool
only migrates into an empty target, to avoid ambiguous merge semantics
(does an existing row win, or the incoming one?) that nothing in the
spec resolves.
Raises
------
MigrationRefusedError
"""
insp = inspect(engine)
with engine.connect() as conn:
for table in TABLES_IN_MIGRATION_ORDER:
if not insp.has_table(table.name):
continue
count = conn.execute(select(func.count()).select_from(table)).scalar()
if count:
raise MigrationRefusedError(
f"target table {table.name!r} already has {count} row(s) "
"-- this tool only migrates into an empty target. Point "
"--to at a fresh database, or clear it first."
)
def _build_reseed_statement(dialect: str, preparer, table_name: str, max_id: int) -> str:
"""Build the reseed statement for a given dialect.
Uses the dialect's identifier_preparer to safely quote the table name,
handling special characters like quotes and brackets.
Parameters
----------
dialect : str
The dialect name ('mssql', 'mysql', etc.)
preparer
The engine's dialect.identifier_preparer
table_name : str
The table name to reseed
max_id : int
The maximum ID to reseed to
Returns
-------
str
The SQL statement text
"""
quoted_table = preparer.quote(table_name)
if dialect == "mssql":
# identifier_preparer.quote() leaves ordinary names unquoted and brackets them
# when required, doubling ] for escaping. DBCC CHECKIDENT accepts an identifier,
# so no string literal is needed.
return f"DBCC CHECKIDENT ({quoted_table}, RESEED, {int(max_id)})"
elif dialect == "mysql":
# identifier_preparer.quote() leaves ordinary names unquoted and backtick-quotes
# them when required, doubling ` for escaping. ALTER TABLE accepts an identifier.
return f"ALTER TABLE {quoted_table} AUTO_INCREMENT = {int(max_id) + 1}"
else:
raise ValueError(f"Unsupported dialect for reseed: {dialect}")
def _reset_autoincrement(conn, engine: Engine, table: Table, pk_column: str) -> None:
"""Reseed *table*'s autoincrement sequence to ``MAX(pk_column) + 1`` on
the target, after inserting rows with explicit primary-key values --
without this, the running application's *next* insert (which supplies
no explicit id) could collide with one this tool just copied in.
Identifiers are safely quoted using the dialect's identifier_preparer
to handle special characters in table names.
"""
max_id = conn.execute(select(func.max(table.c[pk_column]))).scalar()
if max_id is None:
return # the table is empty after the copy -- nothing to reseed
dialect = engine.dialect.name
if dialect == "postgresql":
conn.execute(
text("SELECT setval(pg_get_serial_sequence(:t, :c), :v, true)"),
{"t": table.name, "c": pk_column, "v": max_id},
)
elif dialect == "mssql":
# DBCC CHECKIDENT does not accept the reseed value as a bind parameter.
# Use _build_reseed_statement to safely quote the table name.
stmt = _build_reseed_statement(
dialect, engine.dialect.identifier_preparer, table.name, max_id
)
conn.execute(text(stmt))
elif dialect == "mysql":
# ALTER TABLE accepts identifiers. Use _build_reseed_statement to safely quote.
stmt = _build_reseed_statement(
dialect, engine.dialect.identifier_preparer, table.name, max_id
)
conn.execute(text(stmt))
elif dialect == "sqlite":
# appdb.models declares these columns without Table's
# sqlite_autoincrement=True kwarg, so the DDL SQLAlchemy emits for
# them (see appdb.models's own module docstring) is a plain
# "INTEGER PRIMARY KEY" -- a rowid alias, not a monotonic
# AUTOINCREMENT counter. SQLite computes the next rowid as
# MAX(rowid) + 1 from the table's *actual current contents* on
# every insert that omits one, so there is no persistent counter
# here to desync -- nothing to reset. Kept in step defensively only
# if a sqlite_sequence table exists anyway (a future schema change
# opting into AUTOINCREMENT).
#
# The inspector is built on *conn*, never on the engine. This was
# found the hard way against a target still on ``StaticPool``
# (``appdb.engine.build_engine`` gave every SQLite URL that pool
# before the concurrency fix that split its choice by SQLite
# shape): every checkout shared one DBAPI connection, so
# ``inspect(engine)`` opened a second Connection facade over the
# *same* underlying connection, and closing that facade returned
# it to the pool, which reset it with a ROLLBACK. That rollback
# landed on this transaction -- every row written so far was
# discarded silently, with no exception and no warning, so
# ``import_export`` returned normally having copied nothing.
# ``inspect(conn)`` never opens a second connection at all, so it
# never had this failure mode.
#
# A file-backed target (every ``target_url`` this function is
# ever called with in this codebase) now gets ``QueuePool``
# instead, under which ``inspect(engine)`` would open a genuinely
# separate physical connection rather than reusing this one, so
# the specific hazard above no longer applies to it -- but
# ``inspect(conn)`` is kept regardless, both because it is simply
# the correct tool (this function is already inside a transaction
# on *conn*; there is no reason to open a second connection to
# inspect the same database) and because an in-memory target
# still built through ``appdb.engine.build_engine`` keeps
# ``StaticPool`` (see that function's docstring) and would still
# hit this exact bug.
#
# The per-table verification in :func:`verify_migration` is what
# caught this originally, and is how it was found: every copy step
# reported success and the target was empty.
if inspect(conn).has_table("sqlite_sequence"):
conn.execute(
text("INSERT OR REPLACE INTO sqlite_sequence(name, seq) VALUES (:n, :v)"),
{"n": table.name, "v": max_id},
)
# ---------------------------------------------------------------------------
# Import
# ---------------------------------------------------------------------------
def preview_migration(export: MigrationExport, target_url: str) -> dict[str, int]:
"""Read-only preview for ``--dry-run``: the same refusal checks a real
import would make (schema version, empty target), but never creates a
table or writes a row anywhere. Returns ``{table_name: row_count}`` for
what a real import would copy.
Raises
------
MigrationRefusedError
"""
check_schema_version(export)
engine = build_engine(target_url)
try:
check_target_is_empty(engine)
finally:
engine.dispose()
return {name: snapshot.row_count for name, snapshot in export.tables.items()}
def import_export(export: MigrationExport, target_url: str) -> None:
"""Write *export* into *target_url*: create tables if needed (idempotent,
same as the zero-configuration path -- ``docs/admin-panel-architecture.md``
§5.3), refuse if the target is not empty, then insert every row with its
original primary key and reseed the two autoincrement sequences.
Raises
------
MigrationRefusedError
"""
check_schema_version(export)
engine = build_engine(target_url)
try:
metadata.create_all(engine, checkfirst=True)
check_target_is_empty(engine)
with engine.begin() as conn:
for table in TABLES_IN_MIGRATION_ORDER:
snapshot = export.tables[table.name]
if not snapshot.rows:
continue
conn.execute(table.insert(), [dict(r) for r in snapshot.rows])
if table.name in AUTOINCREMENT_PK_COLUMNS:
_reset_autoincrement(
conn, engine, table, AUTOINCREMENT_PK_COLUMNS[table.name]
)
finally:
engine.dispose()
# ---------------------------------------------------------------------------
# Verification (§5) -- independent of the copy step above
# ---------------------------------------------------------------------------
def verify_migration(source_url: str, target_url: str) -> VerificationReport:
"""Re-read both databases from scratch and compare, per table, row
counts and content hash. Independent of :func:`import_export`: this
reads both sides fresh, so it catches a target corrupted *after* a
successful copy just as readily as an incomplete one.
"""
source_engine = build_engine(source_url)
target_engine = build_engine(target_url)
try:
source_tables = _snapshot_all_tables(source_engine)
target_tables = _snapshot_all_tables(target_engine)
finally:
source_engine.dispose()
target_engine.dispose()
results = tuple(
TableVerification(
table=table.name,
source_row_count=source_tables[table.name].row_count,
target_row_count=target_tables[table.name].row_count,
source_hash=source_tables[table.name].content_hash,
target_hash=target_tables[table.name].content_hash,
)
for table in TABLES_IN_MIGRATION_ORDER
)
return VerificationReport(tables=results)
# ---------------------------------------------------------------------------
# Orchestration
# ---------------------------------------------------------------------------
def run_migration(
source_url: str,
target_url: str,
*,
dry_run: bool = False,
on_export: Callable[[MigrationExport], None] | None = None,
) -> MigrationResult:
"""Migrate the application database from *source_url* to *target_url*.
Always returns a :class:`MigrationResult` -- every refusal this module
recognises (recent write activity, a schema-version mismatch, a
non-empty target, source and target naming the same database) sets
``.ok = False`` with an explanatory ``.message`` rather than raising, so
a caller (the CLI, or a test) checks one flag instead of catching
exceptions for expected outcomes. ``source_hash_before``/
``source_hash_after`` are always computed, including when this call
fails, to prove the source was never mutated (§6).
*on_export*, if given, is called once with the freshly read
:class:`MigrationExport` for a real (non-dry-run) migration, after the
quiescence check has passed and before anything is written to the
target -- ``scripts/migrate_app_db.py`` uses this to persist "the
exported artefact" (§9) to a temporary file at exactly the point it
exists, rather than reading the source a second time to do so. Never
called for a dry run, which writes nothing anywhere.
A genuinely unexpected error (e.g. the target is unreachable) still
propagates as whatever SQLAlchemy raises.
"""
hash_before = hash_database(source_url)
ok = True
message = ""
schema_version = ""
verification: VerificationReport | None = None
export_row_counts: dict[str, int] = {}
try:
check_not_same_database(source_url, target_url)
export = export_database(source_url)
schema_version = export.schema_version
if dry_run:
export_row_counts = preview_migration(export, target_url)
message = "dry run: no rows written to the target"
else:
if on_export is not None:
on_export(export)
import_export(export, target_url)
export_row_counts = {
name: snapshot.row_count for name, snapshot in export.tables.items()
}
verification = verify_migration(source_url, target_url)
if verification.ok:
message = (
"migration verified: every table's row count and "
"content hash match on both sides"
)
else:
ok = False
message = (
"verification FAILED after copy -- the target must not "
"be used; see the per-table report"
)
except MigrationRefusedError as exc:
ok = False
message = str(exc)
finally:
hash_after = hash_database(source_url)
return MigrationResult(
dry_run=dry_run,
ok=ok,
message=message,
schema_version=schema_version,
source_hash_before=hash_before,
source_hash_after=hash_after,
export_row_counts=export_row_counts,
verification=verification,
)