mafw.db.migration.service

The single internal migration service.

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 SchemaMigrator processor — are thin adapters that delegate to this service and present the returned 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 FAILURE for genuine faults. In every failure path the target database schema and history are left unchanged from their pre-operation state.

Added in version 2.4.

author:: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)

Classes

`MigrationContext`(database, ...)	The resolved operating context for a migration operation.
`MigrationManager`(context)	The single migration service shared by the CLI and the SchemaMigrator.

class mafw.db.migration.service.MigrationContext(database: ~peewee.Database, migration_directory: ~pathlib.Path, registry: ~mafw.db.model_register.ModelRegister = <mafw.db.model_register.ModelRegister object>)[source]

Bases: object

The resolved operating context for a migration operation.

Parameters:

database (peewee.Database) – The already-resolved live database (or an initialised proxy).
migration_directory (Path) – The resolved absolute migration directory.
registry (ModelRegister) – The model register to discover from. Defaults to the global mafw_model_register.

Added in version 2.4.

class mafw.db.migration.service.MigrationManager(context: MigrationContext)[source]

Bases: object

The single migration service shared by the CLI and the SchemaMigrator.

Each public operation (diff(), generate(), status(), up(), baseline()) returns a MigrationOutcome.

Parameters:: context (MigrationContext) – The resolved operating context.

Added in version 2.4.

_compute_diff(models: tuple[type[MAFwBaseModel], ...]) → SchemaDiff[source]

Compute the schema diff, ignoring the internal history table.

Parameters:: models (tuple[type[MAFwBaseModel], ...]) – The discovered models to compare.
Returns:: The computed schema diff.
Return type:: SchemaDiff

_discover() → DiscoveryResult[source]

Run model discovery over the configured register.

Returns:: The deduplicated discovery result.
Return type:: DiscoveryResult

_next_version(scripts: list[MigrationScript]) → str[source]

Return the next zero-padded sortable version identifier.

Parameters:: scripts (list[MigrationScript]) – The existing scripts in the directory.
Returns:: A new version string one greater than the current maximum.
Return type:: str

_validate_directory() → Path[source]

Validate the resolved directory before any write or read of scripts.

Returns:: The validated directory.
Return type:: Path
Raises:: MigrationUnresolvedDirectoryError – when the directory is invalid.

baseline() → MigrationOutcome[source]

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.

Returns:: The baseline outcome.
Return type:: MigrationOutcome

diff() → MigrationOutcome[source]

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.

Returns:: The diff outcome.
Return type:: MigrationOutcome

generate() → MigrationOutcome[source]

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.

Returns:: The generation outcome.
Return type:: MigrationOutcome

status() → MigrationOutcome[source]

Report the applied and pending migrations for the target database.

Returns:: The status outcome.
Return type:: MigrationOutcome

up() → MigrationOutcome[source]

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.

Returns:: The application outcome.
Return type:: MigrationOutcome

property _database: Database: The live database for this operation.

property _directory: Path: The resolved migration directory for this operation.

mafw.db.migration.service._render_baseline_script(version: str, models: tuple[type[MAFwBaseModel], ...]) → str[source]

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.

Parameters:

version (str) – The baseline version identifier.
models (tuple[type[MAFwBaseModel], ...]) – The discovered models accounted for by the baseline.

Returns:

The baseline script body as text.

Return type:

str

mafw.db.migration.service._wrap(exc: Exception) → MigrationError[source]

Wrap a non-migration exception as a migration error.

Parameters:: exc (Exception) – The original exception.
Returns:: A migration error carrying the original message.
Return type:: MigrationError