# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Per-database migration history.
The migration history is a dedicated table stored *inside each target database*,
so history is naturally per-database: the same set of migration scripts applied
to two different databases maintains two independent histories.
The history model is created with ``do_not_register=True`` so it never enters the
global model register and is therefore never itself a migration subject. The
reader/writer helpers record a script only after it applies successfully and
expose consistency checks that compare the recorded name/checksum against the
scripts present in the migration directory.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import datetime
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
import peewee
from mafw.db.db_model import MAFwBaseModel
from mafw.db.migration.errors import MigrationHistoryInconsistencyError
from mafw.db.migration.script import MigrationScript, load_script
#: The fixed table name of the per-database migration history.
MIGRATION_HISTORY_TABLE = 'mafw_migration_history'
"""The fixed table name of the per-database migration history."""
[docs]
class MigrationHistory(MAFwBaseModel, do_not_register=True):
"""
The per-database applied-migration history table.
Created with ``do_not_register=True`` so it never enters
:data:`~mafw.db.db_model.mafw_model_register` and is excluded from discovery
and diffing.
.. versionadded:: 2.4
"""
version = peewee.CharField(primary_key=True, help_text='The migration script version identifier.')
"""The migration script version identifier (primary key)."""
name = peewee.CharField(help_text='The migration script file stem.')
"""The migration script file stem."""
checksum = peewee.CharField(help_text='The script content checksum recorded at apply time.')
"""The script content checksum recorded when the script was applied."""
applied_at = peewee.DateTimeField(help_text='The timestamp at which the script was applied.')
"""The timestamp at which the script was applied."""
class Meta:
"""Metadata pinning the table name and disabling automatic creation."""
table_name = MIGRATION_HISTORY_TABLE
automatic_creation = False
[docs]
@dataclass(frozen=True)
class MigrationHistoryEntry:
"""
A read-only view of a single migration-history row.
:param version: The migration script version identifier.
:type version: str
:param name: The migration script file stem.
:type name: str
:param checksum: The recorded content checksum.
:type checksum: str
:param applied_at: The application timestamp.
:type applied_at: datetime.datetime
.. versionadded:: 2.4
"""
version: str
name: str
checksum: str
applied_at: datetime.datetime
[docs]
def history_table_exists(database: peewee.Database) -> bool:
"""
Return whether the migration-history table exists in the database.
:param database: The database to inspect.
:type database: peewee.Database
:return: ``True`` when the history table is present.
:rtype: bool
"""
return database.table_exists(MIGRATION_HISTORY_TABLE)
[docs]
def ensure_history_table(database: peewee.Database) -> None:
"""
Create the migration-history table if it does not yet exist.
:param database: The database in which to create the table.
:type database: peewee.Database
"""
with database.bind_ctx([MigrationHistory]):
database.create_tables([MigrationHistory], safe=True)
[docs]
def read_history(database: peewee.Database) -> list[MigrationHistoryEntry]:
"""
Read the applied-migration history in chronological order.
:param database: The database to read from.
:type database: peewee.Database
:return: The applied entries ordered by application timestamp then version.
:rtype: list[MigrationHistoryEntry]
"""
if not history_table_exists(database):
return []
with database.bind_ctx([MigrationHistory]):
rows = MigrationHistory.select().order_by(MigrationHistory.applied_at, MigrationHistory.version).execute()
return [
MigrationHistoryEntry(
version=row.version,
name=row.name,
checksum=row.checksum,
applied_at=row.applied_at,
)
for row in rows
]
[docs]
def record_applied(database: peewee.Database, script: MigrationScript) -> MigrationHistoryEntry:
"""
Record a successfully applied script in the history.
:param database: The database whose history to update.
:type database: peewee.Database
:param script: The script that was applied.
:type script: MigrationScript
:return: The recorded history entry.
:rtype: MigrationHistoryEntry
"""
ensure_history_table(database)
applied_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
with database.bind_ctx([MigrationHistory]):
MigrationHistory.insert(
version=script.version,
name=script.name,
checksum=script.checksum,
applied_at=applied_at,
).execute()
return MigrationHistoryEntry(
version=script.version,
name=script.name,
checksum=script.checksum,
applied_at=applied_at,
)
[docs]
def list_scripts(directory: Path) -> list[MigrationScript]:
"""
List the migration scripts in a directory, ordered by version.
:param directory: The migration directory to scan.
:type directory: Path
:return: The scripts found, ordered by their version identifier.
:rtype: list[MigrationScript]
"""
scripts: list[MigrationScript] = []
for path in sorted(directory.glob('*.py')):
if path.name == '__init__.py':
continue
scripts.append(load_script(path))
scripts.sort(key=lambda script: script.version)
return scripts
[docs]
def check_consistency(
entries: Sequence[MigrationHistoryEntry],
scripts: Sequence[MigrationScript],
) -> None:
"""
Verify that the history and the directory are mutually consistent.
A history entry must have a corresponding script in the directory with an
unchanged checksum. A mismatch indicates a missing or altered script.
:param entries: The applied-history entries.
:type entries: Sequence[MigrationHistoryEntry]
:param scripts: The scripts present in the directory.
:type scripts: Sequence[MigrationScript]
:raises MigrationHistoryInconsistencyError: when the state is inconsistent.
"""
by_version: Mapping[str, MigrationScript] = {script.version: script for script in scripts}
missing: list[str] = []
changed: list[str] = []
for entry in entries:
script = by_version.get(entry.version)
if script is None:
missing.append(entry.version)
continue
if script.checksum != entry.checksum:
changed.append(entry.version)
if missing or changed:
parts: list[str] = []
if missing:
parts.append('missing scripts for recorded versions: ' + ', '.join(sorted(missing)))
if changed:
parts.append('changed content for recorded versions: ' + ', '.join(sorted(changed)))
message = (
'The migration history is inconsistent with the migration directory ('
+ '; '.join(parts)
+ '). Restore the affected migration scripts or correct the history before proceeding.'
)
raise MigrationHistoryInconsistencyError(message, scripts=sorted(set(missing) | set(changed)))