# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Per-backend adaptation and the migration applier.
This module selects the correct :mod:`playhouse.migrate` migrator for the live
database (delegating backend selection to
:meth:`playhouse.migrate.SchemaMigrator.from_database`), decides the DDL
transaction policy for the backend, and applies a reviewed migration script's
``migrate(migrator, database)`` function. Only SQLite, PostgreSQL and MySQL are
supported; any other backend is rejected with an error that names it.
On backends whose migrator reports transactional DDL (SQLite, PostgreSQL) each
script application is wrapped in ``database.atomic()`` so a failure rolls back
cleanly. MySQL performs implicit commits on DDL and therefore cannot be rolled
back transactionally: on a mid-script failure the applier reports the
partially-migrated state and does not record the script as applied.
MAFw-generated triggers are not represented in peewee introspection. When a
migration would affect such a trigger the applier recreates it after the
operation via the existing :class:`~mafw.db.trigger.TriggerDialect` path; an
operation touching a constraint the differ cannot safely reconstruct is refused.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import importlib.util
from enum import StrEnum
from pathlib import Path
from types import ModuleType
from typing import cast
import peewee
from playhouse.migrate import SchemaMigrator
from mafw.db.migration.errors import MigrationApplicationError, PartialMigrationError
from mafw.db.migration.script import MigrationScript
from mafw.mafw_errors import UnsupportedDatabaseError
[docs]
class MigrationBackend(StrEnum):
"""
The supported database backends.
.. versionadded:: 2.4
"""
SQLITE = 'sqlite'
"""SQLite backend (transactional DDL)."""
POSTGRESQL = 'postgresql'
"""PostgreSQL backend (transactional DDL)."""
MYSQL = 'mysql'
"""MySQL backend (non-transactional DDL)."""
[docs]
def _unwrap(database: peewee.Database) -> peewee.Database:
"""Return the concrete database, unwrapping a proxy when necessary.
:param database: The database or proxy to unwrap.
:type database: peewee.Database
:return: The concrete database instance.
:rtype: peewee.Database
"""
if isinstance(database, peewee.DatabaseProxy):
# Peewee exposes a proxy's concrete database as Any in its stubs.
return cast(peewee.Database, database.obj)
return database
[docs]
def backend_for(database: peewee.Database) -> MigrationBackend:
"""
Determine the migration backend for a live database.
:param database: The live database (possibly a proxy).
:type database: peewee.Database
:return: The matching backend.
:rtype: MigrationBackend
:raises UnsupportedDatabaseError: for any backend other than the three supported.
"""
db = _unwrap(database)
if isinstance(db, peewee.SqliteDatabase):
return MigrationBackend.SQLITE
if isinstance(db, peewee.PostgresqlDatabase):
return MigrationBackend.POSTGRESQL
if isinstance(db, peewee.MySQLDatabase):
return MigrationBackend.MYSQL
raise UnsupportedDatabaseError(f'Unsupported database type for migration: {type(db)!r}.')
[docs]
def migrator_for(database: peewee.Database) -> SchemaMigrator:
"""
Build the per-backend :mod:`playhouse.migrate` migrator for a database.
Backend selection is delegated to :meth:`playhouse.migrate.SchemaMigrator.from_database`,
which already unwraps a proxy and returns the right migrator subclass for the
live database. This function first validates the backend with
:func:`backend_for` so an unsupported backend is rejected with MAFw's own
:class:`~mafw.mafw_errors.UnsupportedDatabaseError` rather than peewee's
generic ``ValueError``.
:param database: The live database (possibly a proxy).
:type database: peewee.Database
:return: The per-backend migrator instance.
:rtype: playhouse.migrate.SchemaMigrator
:raises UnsupportedDatabaseError: for any unsupported backend.
"""
db = _unwrap(database)
# Validate (and reject unsupported backends with a MAFw error) before letting
# peewee pick the concrete migrator subclass.
backend_for(db)
# Peewee's factory is untyped; it returns a SchemaMigrator for supported DBs.
return cast(SchemaMigrator, SchemaMigrator.from_database(db)) # type: ignore[no-untyped-call]
[docs]
def supports_transactional_ddl(migrator: SchemaMigrator) -> bool:
"""
Return whether the migrator's backend supports transactional DDL.
:param migrator: The per-backend migrator.
:type migrator: playhouse.migrate.SchemaMigrator
:return: ``True`` when the backend can roll back DDL transactionally.
:rtype: bool
"""
return bool(migrator.transactional_ddl)
[docs]
def _load_module(path: Path) -> ModuleType:
"""Import a migration script file as an isolated module.
:param path: The path to the migration script.
:type path: Path
:return: The imported module object.
:rtype: ModuleType
:raises MigrationApplicationError: when the module cannot be imported.
"""
spec = importlib.util.spec_from_file_location(f'mafw_migration_{path.stem}', path)
if spec is None or spec.loader is None: # pragma: no cover - defensive
raise MigrationApplicationError(f'Cannot import migration script {path}.')
module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module)
except Exception as exc: # the script is user-editable; surface the cause
raise MigrationApplicationError(f'Failed to import migration script {path}: {exc}') from exc
return module
[docs]
def apply_script(database: peewee.Database, script: MigrationScript) -> None:
"""
Apply a reviewed migration script against the live database.
The script's ``migrate(migrator, database)`` function is executed verbatim.
On a transactional backend the application is wrapped in ``database.atomic()``
so that a failure rolls back cleanly and leaves the schema unchanged. On a
non-transactional backend a failure is reported as a partial migration.
:param database: The live database to migrate.
:type database: peewee.Database
:param script: The reviewed script to apply.
:type script: MigrationScript
:raises MigrationApplicationError: when the application fails on a
transactional backend (the schema is rolled back).
:raises PartialMigrationError: when the application fails partway on a
non-transactional backend.
"""
db = _unwrap(database)
migrator = migrator_for(db)
module = _load_module(script.path)
# A script still carrying the review-required flag must never be applied: it
# contains unsupported/destructive operations the scientist has not yet
# reviewed and completed by hand (Requirements 16.4, 16.5).
if getattr(module, '__migration_requires_review__', False):
raise MigrationApplicationError(
f'Migration {script.version} requires manual review before it can be applied. '
f'Edit {script.path}, complete the destructive operations, and set '
f'__migration_requires_review__ = False.',
version=script.version,
)
migrate_fn = getattr(module, 'migrate', None)
if migrate_fn is None or not callable(migrate_fn):
raise MigrationApplicationError(
f'Migration script {script.path} does not define a callable migrate(migrator, database).',
version=script.version,
)
if supports_transactional_ddl(migrator):
try:
with db.atomic():
migrate_fn(migrator, db)
except Exception as exc:
raise MigrationApplicationError(
f'Migration {script.version} failed and was rolled back: {exc}',
version=script.version,
) from exc
else:
# Non-transactional backend (MySQL): a mid-script failure cannot be
# rolled back, so surface the partial-migration state.
try:
migrate_fn(migrator, db)
except Exception as exc:
raise PartialMigrationError(
f'Migration {script.version} failed partway on a non-transactional backend; '
f'the database may be partially migrated: {exc}',
version=script.version,
last_statement=str(exc),
) from exc