# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
The single internal migration service.
:class:`MigrationManager` is the one component that performs model discovery,
schema diffing, script generation, status reporting, history application and
baselining. Both front ends — the ``mafw db`` CLI command group and the
:class:`~mafw.processor_library.db_init.SchemaMigrator` processor — are thin
adapters that delegate to this service and present the returned
:class:`~mafw.db.migration.outcome.MigrationOutcome`. No front end performs any
diff, generation, status computation or application itself, which guarantees that
both entry points agree for identical inputs.
The service converts *expected* conditions (an empty diff, a pending conflict, an
unsupported-only diff, a legacy database) into outcomes and reserves
:attr:`~mafw.db.migration.outcome.MigrationResult.FAILURE` for genuine faults. In
every failure path the target database schema and history are left unchanged from
their pre-operation state.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
import peewee
from mafw.db.db_model import MAFwBaseModel, mafw_model_register
from mafw.db.migration import backend as backend_mod
from mafw.db.migration.diff import SchemaDiff, compute_diff
from mafw.db.migration.directory import database_lock, directory_lock, validate_directory
from mafw.db.migration.discovery import DiscoveryResult, discover
from mafw.db.migration.errors import (
MigrationApplicationError,
MigrationConfigurationError,
MigrationConflictingTableNameError,
MigrationError,
MigrationHistoryInconsistencyError,
MigrationUnresolvedDirectoryError,
)
from mafw.db.migration.history import (
MIGRATION_HISTORY_TABLE,
MigrationHistoryEntry,
check_consistency,
ensure_history_table,
history_table_exists,
list_scripts,
read_history,
record_applied,
)
from mafw.db.migration.outcome import MigrationOutcome, MigrationResult, failure, make_outcome
from mafw.db.migration.script import (
MigrationScript,
render_script,
script_name,
write_script,
)
from mafw.db.model_register import ModelRegister
from mafw.mafw_errors import UnsupportedDatabaseError
[docs]
@dataclass(frozen=True)
class MigrationContext:
"""
The resolved operating context for a migration operation.
:param database: The already-resolved live database (or an initialised proxy).
:type database: peewee.Database
:param migration_directory: The resolved absolute migration directory.
:type migration_directory: Path
:param registry: The model register to discover from. Defaults to the global
:data:`~mafw.db.db_model.mafw_model_register`.
:type registry: ModelRegister
.. versionadded:: 2.4
"""
database: peewee.Database
migration_directory: Path
registry: ModelRegister = field(default=mafw_model_register)
[docs]
class MigrationManager:
"""
The single migration service shared by the CLI and the SchemaMigrator.
Each public operation (:meth:`diff`, :meth:`generate`, :meth:`status`,
:meth:`up`, :meth:`baseline`) returns a
:class:`~mafw.db.migration.outcome.MigrationOutcome`.
:param context: The resolved operating context.
:type context: MigrationContext
.. versionadded:: 2.4
"""
def __init__(self, context: MigrationContext) -> None:
self.context = context
# -- internal helpers -------------------------------------------------
@property
def _database(self) -> peewee.Database:
"""The live database for this operation."""
return self.context.database
@property
def _directory(self) -> Path:
"""The resolved migration directory for this operation."""
return self.context.migration_directory
[docs]
def _discover(self) -> DiscoveryResult:
"""Run model discovery over the configured register.
:return: The deduplicated discovery result.
:rtype: DiscoveryResult
"""
return discover(self.context.registry)
[docs]
def _compute_diff(self, models: tuple[type[MAFwBaseModel], ...]) -> SchemaDiff:
"""Compute the schema diff, ignoring the internal history table.
:param models: The discovered models to compare.
:type models: tuple[type[MAFwBaseModel], ...]
:return: The computed schema diff.
:rtype: SchemaDiff
"""
return compute_diff(models, self._database, ignore_tables=frozenset({MIGRATION_HISTORY_TABLE}))
[docs]
def _validate_directory(self) -> Path:
"""Validate the resolved directory before any write or read of scripts.
:return: The validated directory.
:rtype: Path
:raises MigrationUnresolvedDirectoryError: when the directory is invalid.
"""
return validate_directory(self._directory)
[docs]
def _next_version(self, scripts: list[MigrationScript]) -> str:
"""Return the next zero-padded sortable version identifier.
:param scripts: The existing scripts in the directory.
:type scripts: list[MigrationScript]
:return: A new version string one greater than the current maximum.
:rtype: str
"""
numbers = [int(script.version) for script in scripts if script.version.isdigit()]
return f'{(max(numbers) + 1) if numbers else 1:04d}'
# -- operations -------------------------------------------------------
[docs]
def diff(self) -> MigrationOutcome:
"""
Report the differences between the discovered models and the database.
This operation is strictly read-only: it modifies neither the schema, nor
the data, nor any migration script. An empty diff is reported as scoped
synchronization only, never as full schema equivalence.
:return: The diff outcome.
:rtype: MigrationOutcome
"""
try:
discovery = self._discover()
diff = self._compute_diff(discovery.models)
except (MigrationConflictingTableNameError, UnsupportedDatabaseError) as exc:
return failure(exc if isinstance(exc, MigrationError) else _wrap(exc))
except peewee.PeeweeException as exc:
return failure(MigrationConfigurationError(f'The target database is unavailable: {exc}'))
if diff.is_empty:
return make_outcome(
MigrationResult.NO_CHANGE,
diff=diff,
scoped_equivalence_only=True,
messages=(
'No difference detected within the documented supported difference types. '
'This does not guarantee full schema equivalence.',
),
)
if not diff.has_supported:
return make_outcome(
MigrationResult.UNSUPPORTED_OR_AMBIGUOUS,
diff=diff,
messages=tuple(f'Unsupported/ambiguous: {d.describe()}' for d in diff.unsupported),
)
return make_outcome(
MigrationResult.DIFFERENCE_DETECTED,
diff=diff,
messages=tuple(d.describe() for d in diff.all_differences),
)
[docs]
def generate(self) -> MigrationOutcome:
"""
Generate one reviewable migration script for the detected diff.
The operation never alters the database schema or contents. If a pending
migration already matches the diff it is reported instead of generating a
duplicate.
When the diff contains unsupported/destructive differences, a script is
still generated so the scientist has a concrete starting point: the
supported (additive) operations are emitted as executable code, and each
unsupported difference becomes a loud, commented review stub. Such a
script carries ``__migration_requires_review__ = True`` and the applier
refuses it until the scientist has completed the operations by hand and
cleared the flag.
:return: The generation outcome.
:rtype: MigrationOutcome
"""
try:
self._validate_directory()
discovery = self._discover()
diff = self._compute_diff(discovery.models)
except MigrationError as exc:
return failure(exc)
except UnsupportedDatabaseError as exc:
return failure(_wrap(exc))
except peewee.PeeweeException as exc:
return failure(MigrationConfigurationError(f'The target database is unavailable: {exc}'))
if diff.is_empty:
return make_outcome(
MigrationResult.NO_CHANGE,
diff=diff,
scoped_equivalence_only=True,
messages=('No supported difference to generate a migration for.',),
)
# Only a *pending* migration — one present in the directory but not yet
# recorded as applied — can block a new generation. An already-applied
# script must never block a later, logically-equivalent change: the
# same diff (e.g. re-adding a column that was previously dropped) yields
# the same content digest, so matching against the whole directory would
# wrongly report PENDING. We therefore compare the digest only against the
# scripts that are still pending for this database.
existing = list_scripts(self._directory)
try:
applied_versions = {entry.version for entry in read_history(self._database)}
except peewee.PeeweeException as exc:
return failure(MigrationConfigurationError(f'The target database is unavailable: {exc}'))
pending_scripts = [script for script in existing if script.version not in applied_versions]
version = self._next_version(existing)
name = script_name(diff, version)
target_digest = name.split('_', 1)[1]
for script in pending_scripts:
# The digest embedded in the name identifies the diff content.
if script.name.endswith(target_digest):
return make_outcome(
MigrationResult.PENDING,
diff=diff,
scripts=(script,),
pending=(script,),
messages=(f'A pending migration already exists for this change: {script.name}.',),
)
try:
with directory_lock(self._directory):
body = render_script(diff, version)
script = write_script(self._directory, name, body)
except MigrationError as exc:
return failure(exc)
messages = [f'Generated migration script: {script.path}']
if diff.unsupported:
messages.append(
'This migration contains unsupported/destructive change(s) that need manual review: '
+ '; '.join(d.describe() for d in diff.unsupported)
)
messages.append(
'The script flags them as review-required and will NOT be applied until you edit it, '
'complete the operations by hand, and set __migration_requires_review__ = False.'
)
return make_outcome(
MigrationResult.GENERATED,
diff=diff,
scripts=(script,),
messages=tuple(messages),
)
[docs]
def status(self) -> MigrationOutcome:
"""
Report the applied and pending migrations for the target database.
:return: The status outcome.
:rtype: MigrationOutcome
"""
try:
self._validate_directory()
except MigrationUnresolvedDirectoryError as exc:
return failure(exc)
try:
scripts = list_scripts(self._directory)
except ValueError as exc:
return failure(MigrationHistoryInconsistencyError(str(exc)))
try:
legacy = not history_table_exists(self._database)
entries = read_history(self._database)
except peewee.PeeweeException as exc:
return failure(MigrationConfigurationError(f'The target database is unavailable: {exc}'))
try:
check_consistency(entries, scripts)
except MigrationHistoryInconsistencyError as exc:
return failure(exc)
applied_versions = {entry.version for entry in entries}
pending = [script for script in scripts if script.version not in applied_versions]
if legacy:
messages = (
'This database has no migration history (legacy database). '
'Run "mafw db migrate baseline" to start tracking migrations.',
)
else:
messages = (f'{len(entries)} applied migration(s); {len(pending)} pending migration(s).',)
return make_outcome(
MigrationResult.PENDING if pending else MigrationResult.NO_CHANGE,
applied=entries,
pending=pending,
messages=messages,
legacy=legacy,
)
[docs]
def up(self) -> MigrationOutcome:
"""
Apply all pending migrations in ascending version order.
Each script is applied exactly once; a successfully applied script is
recorded in the history before the next is applied. On the first failure
the operation halts, leaving the schema and history consistent, and
reports the failing script by version. After all pending migrations are
applied, the resulting schema is re-checked against the models.
:return: The application outcome.
:rtype: MigrationOutcome
"""
try:
self._validate_directory()
scripts = list_scripts(self._directory)
discovery = self._discover()
except MigrationError as exc:
return failure(exc)
except ValueError as exc:
return failure(MigrationHistoryInconsistencyError(str(exc)))
try:
entries = read_history(self._database)
check_consistency(entries, scripts)
except MigrationHistoryInconsistencyError as exc:
return failure(exc)
except peewee.PeeweeException as exc:
return failure(MigrationConfigurationError(f'The target database is unavailable: {exc}'))
applied_versions = {entry.version for entry in entries}
pending = [script for script in scripts if script.version not in applied_versions]
pending.sort(key=lambda script: script.version)
if not pending:
return make_outcome(
MigrationResult.NO_CHANGE,
applied=entries,
messages=('The database is already up to date.',),
)
# Refuse up front if any pending script still requires manual review, so
# the operator gets a clear, actionable message before anything is applied.
review_required = [script for script in pending if script.requires_review]
if review_required:
names = ', '.join(script.name for script in review_required)
return failure(
MigrationApplicationError(
f'{len(review_required)} pending migration(s) require manual review before they can be '
f'applied: {names}.',
version=review_required[0].version,
),
messages=(
f'The following migration(s) contain unsupported/destructive operations and are flagged '
f'as review-required: {names}.',
'Edit each script, complete the operations by hand, and set '
'__migration_requires_review__ = False before running "mafw db migrate up" again.',
),
)
newly_applied: list[MigrationHistoryEntry] = []
try:
ensure_history_table(self._database)
with database_lock(self._database):
for script in pending:
backend_mod.apply_script(self._database, script)
entry = record_applied(self._database, script)
newly_applied.append(entry)
except MigrationApplicationError as exc:
return failure(
exc,
messages=(
f'Migration {exc.version} failed: {exc}',
f'{len(newly_applied)} migration(s) applied before the failure.',
),
)
# Re-check the schema against the models after applying everything.
post_diff = self._compute_diff(discovery.models)
messages = [f'Applied {len(newly_applied)} migration(s).']
if post_diff.is_empty:
messages.append('The schema is compatible with the discovered models (scoped to supported types).')
else:
messages.append('The schema is NOT fully compatible; remaining differences:')
messages.extend(f' - {d.describe()}' for d in post_diff.all_differences)
return make_outcome(
MigrationResult.APPLIED,
diff=post_diff,
applied=tuple(read_history(self._database)),
messages=tuple(messages),
)
[docs]
def baseline(self) -> MigrationOutcome:
"""
Baseline a legacy database by recording its existing schema.
On a legacy database lacking a history, this creates the history and
records the existing schema as the initial applied entry without executing
any create/alter/drop against the existing tables. It refuses (without
modification) when a history already exists, and it aborts without
changing anything when it cannot account for every discovered model.
:return: The baseline outcome.
:rtype: MigrationOutcome
"""
try:
self._validate_directory()
discovery = self._discover()
except MigrationError as exc:
return failure(exc)
except UnsupportedDatabaseError as exc:
return failure(_wrap(exc))
try:
if history_table_exists(self._database):
return failure(
MigrationApplicationError('A migration history already exists; baseline refuses to modify it.')
)
# Verify that every discovered model's table exists before recording.
missing = [
model.__name__ for model in discovery.models if not self._database.table_exists(model._meta.table_name)
]
if missing:
return failure(
MigrationApplicationError(
'Baseline cannot account for every discovered model. '
'Missing tables for models: ' + ', '.join(sorted(missing))
)
)
except peewee.PeeweeException as exc:
return failure(MigrationConfigurationError(f'The target database is unavailable: {exc}'))
# Record the baseline as the initial applied entry without any DDL on
# the existing tables. A real baseline script is written to the directory
# (a no-op migrate) so that the history and the directory stay consistent
# and ``status``/``up`` do not report an inconsistency afterwards.
version = '0000'
name = f'{version}_baseline'
body = _render_baseline_script(version, discovery.models)
with database_lock(self._database):
with directory_lock(self._directory):
script = write_script(self._directory, name, body)
ensure_history_table(self._database)
entry = record_applied(self._database, script)
return make_outcome(
MigrationResult.APPLIED,
applied=(entry,),
messages=(
f'Baselined the database: recorded {len(discovery.models)} model(s) '
f'as the initial migration entry {version}.',
),
)
[docs]
def _render_baseline_script(version: str, models: tuple[type[MAFwBaseModel], ...]) -> str:
"""Render a no-op baseline migration script body.
The baseline script records the first-adoption point of a legacy database. It
performs no schema change: its ``migrate`` function is intentionally empty so
that applying it (or re-checking it) never touches existing tables.
:param version: The baseline version identifier.
:type version: str
:param models: The discovered models accounted for by the baseline.
:type models: tuple[type[MAFwBaseModel], ...]
:return: The baseline script body as text.
:rtype: str
"""
table_list = ', '.join(sorted(m._meta.table_name for m in models)) or '(none)'
lines = [
'# Generated by the MAFw schema-migration service (baseline).',
'# This baseline records the existing schema of a legacy database.',
'# It performs no schema change; do not add operations to it.',
'"""Auto-generated MAFw baseline migration script."""',
'from __future__ import annotations',
'',
'import peewee',
'from playhouse.migrate import SchemaMigrator',
'',
f"__migration_version__ = '{version}'",
f'# Baselined tables: {table_list}',
'',
'',
'def migrate(migrator: SchemaMigrator, database: peewee.Database) -> None:',
' """Baseline marker: no schema change is applied."""',
' return None',
'',
]
return '\n'.join(lines)
[docs]
def _wrap(exc: Exception) -> MigrationError:
"""Wrap a non-migration exception as a migration error.
:param exc: The original exception.
:type exc: Exception
:return: A migration error carrying the original message.
:rtype: MigrationError
"""
if isinstance(exc, MigrationError):
return exc
return MigrationApplicationError(str(exc))