# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Runtime helpers used by generated migration scripts.
Generated migration scripts import from this module so that the script body
stays compact and readable for a reviewing scientist. The helpers wrap the
small amount of :mod:`playhouse.migrate` and :mod:`peewee` boilerplate needed to
create a new table or add a nullable column, which are the additive operations
the differ currently supports.
They also expose :func:`temporary_trigger_removal` (a context manager, and its
underlying :func:`remove_all_triggers` / :func:`restore_all_triggers`
primitives), which a reviewed destructive migration should bracket its
operations with on SQLite. Several ``ALTER``-limited backends (SQLite in
particular) implement a column drop/rename/type change by rebuilding the table
(create a temporary table, copy the data, drop the original, rename the temporary
one). The backend refuses to drop the original table while a trigger still points
at it, and MAFw triggers are a MAFw extension that :mod:`playhouse.migrate` knows
nothing about. Dropping every MAFw trigger before the operations and recreating
them afterwards removes that clash; this is exactly what
:class:`~mafw.processor_library.db_init.TriggerRefresher` does, so it is safe. The
context manager restores the triggers in a ``finally`` clause, so they are never
left dropped even if the migration body raises.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import contextlib
import logging
from collections.abc import Iterator
from typing import cast
import peewee
from mafw.db.trigger import MySQLDialect, PostgreSQLDialect, SQLiteDialect, TriggerDialect
log = logging.getLogger(__name__)
#: Mapping from the peewee ``field_type`` token to a nullable field factory.
#: Only the common additive types are mapped; unknown types fall back to a
#: nullable :class:`peewee.TextField`.
_FIELD_FACTORIES: dict[str, type[peewee.Field]] = {
'AUTO': peewee.IntegerField,
'INT': peewee.IntegerField,
'BIGINT': peewee.BigIntegerField,
'SMALLINT': peewee.SmallIntegerField,
'FLOAT': peewee.FloatField,
'DOUBLE': peewee.DoubleField,
'REAL': peewee.FloatField,
'DECIMAL': peewee.DecimalField,
'VARCHAR': peewee.CharField,
'TEXT': peewee.TextField,
'CHAR': peewee.CharField,
'BOOL': peewee.BooleanField,
'DATETIME': peewee.DateTimeField,
'DATE': peewee.DateField,
'TIME': peewee.TimeField,
'BLOB': peewee.BlobField,
}
[docs]
def nullable_field(field_type: str) -> peewee.Field:
"""
Build a nullable peewee field of the requested logical type.
:param field_type: The peewee ``field_type`` token (for example ``VARCHAR``).
:type field_type: str
:return: A nullable field instance suitable for ``migrator.add_column``.
:rtype: peewee.Field
"""
factory = _FIELD_FACTORIES.get(field_type.upper(), peewee.TextField)
return factory(null=True)
[docs]
def create_table(database: peewee.Database, table_name: str) -> None:
"""
Create a table for a model by resolving it from the model register.
The table is created with ``safe=True`` so that an already-existing table
(for example one created by :class:`~mafw.processor_library.db_init.TableCreator`)
is not an error.
:param database: The live database the migration runs against.
:type database: peewee.Database
:param table_name: The name of the table to create.
:type table_name: str
"""
from mafw.db.db_model import mafw_model_register
model = mafw_model_register.get_model(table_name)
with database.bind_ctx([model]): # type: ignore[list-item]
database.create_tables([model], safe=True) # type: ignore[list-item]
[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 _trigger_dialect(database: peewee.Database) -> TriggerDialect:
"""Return the MAFw trigger dialect for the live database.
:param database: The live database (possibly a proxy).
:type database: peewee.Database
:return: The matching trigger dialect.
:rtype: TriggerDialect
:raises UnsupportedDatabaseError: for an unsupported backend.
"""
from mafw.mafw_errors import UnsupportedDatabaseError
db = _unwrap(database)
if isinstance(db, peewee.SqliteDatabase):
return SQLiteDialect()
if isinstance(db, peewee.MySQLDatabase):
return MySQLDialect()
if isinstance(db, peewee.PostgresqlDatabase):
return PostgreSQLDialect()
raise UnsupportedDatabaseError(f'Unsupported database type for trigger handling: {type(db)!r}.')
[docs]
def remove_all_triggers(database: peewee.Database) -> list[tuple[str, str]]:
"""
Drop every database trigger and return the affected ``(trigger, table)`` pairs.
Call this inside a reviewed destructive migration *before* the
``migrator.migrate(...)`` operations, and pair it with
:func:`restore_all_triggers` afterwards. It removes the clash between a
table-rebuild operation (common on SQLite, which refuses to drop a table a
trigger points at) and MAFw's triggers, which :mod:`playhouse.migrate` does
not manage. Dropping and recreating MAFw triggers is safe because they are
regenerable from the model register, mirroring
:class:`~mafw.processor_library.db_init.TriggerRefresher`.
:param database: The live database to operate on.
:type database: peewee.Database
:return: The list of ``(trigger_name, table_name)`` pairs that were dropped.
:rtype: list[tuple[str, str]]
"""
db = _unwrap(database)
dialect = _trigger_dialect(db)
rows: list[tuple[str, str]] = db.execute_sql(dialect.select_all_trigger_sql()).fetchall() # type: ignore[no-untyped-call]
for trigger_name, table_name in rows:
db.execute_sql(dialect.drop_trigger_sql(trigger_name, safe=True, table_name=table_name)) # type: ignore[no-untyped-call]
if rows:
log.info('Temporarily dropped %d trigger(s) for the migration.', len(rows))
return [(str(trigger), str(table)) for trigger, table in rows]
[docs]
def restore_all_triggers(database: peewee.Database) -> None:
"""
Recreate every MAFw trigger after a destructive migration.
Call this inside a reviewed destructive migration *after* the
``migrator.migrate(...)`` operations, to recreate the triggers that
:func:`remove_all_triggers` dropped. Triggers are recreated by re-running the
(idempotent, ``safe=True``) table creation for every registered model, which
regenerates the model's declared triggers via
:meth:`~mafw.db.db_model.MAFwBaseModel.create_table` without touching
existing data.
:param database: The live database to operate on.
:type database: peewee.Database
"""
from mafw.db.db_model import mafw_model_register
db = _unwrap(database)
models = [model for _, model in mafw_model_register.items()]
if not models:
return
with db.bind_ctx(models): # type: ignore[arg-type]
db.create_tables(models, safe=True) # type: ignore[arg-type]
log.info('Recreated MAFw triggers after the migration.')
[docs]
@contextlib.contextmanager
def temporary_trigger_removal(database: peewee.Database) -> Iterator[list[tuple[str, str]]]:
"""
Context manager that drops MAFw triggers for the duration of a migration.
Enter drops every database trigger (via :func:`remove_all_triggers`) and
yields the dropped ``(trigger, table)`` pairs; exit recreates every
registered model's triggers (via :func:`restore_all_triggers`) in a
``finally`` clause, so the triggers are **always** restored — on normal exit
*and* when the body raises. This is the recommended way to bracket a reviewed
destructive migration on an ``ALTER``-limited backend (SQLite), where a table
rebuild fails while a trigger still points at the table.
.. note::
The MAFw applier runs the whole migration inside a single
``database.atomic()`` on transactional backends (SQLite, PostgreSQL); the
per-operation ``playhouse.migrate`` runner does **not** open its own
transaction. On those backends a failure therefore rolls back the trigger
drops together with the schema change, and this context manager's
``finally`` recreation runs within the same (about-to-roll-back)
transaction, which is harmless. On a non-transactional backend (MySQL) the
``finally`` recreation is what actually guarantees the triggers are not
left dropped after a mid-migration failure.
Example inside a generated migration script::
def migrate(migrator, database):
operations = [migrator.drop_column('widget', 'gone')]
with temporary_trigger_removal(database):
migrator.migrate(*operations)
:param database: The live database to operate on.
:type database: peewee.Database
:return: A context manager yielding the list of dropped ``(trigger, table)``
pairs.
:rtype: Iterator[list[tuple[str, str]]]
"""
db = _unwrap(database)
removed = remove_all_triggers(db)
try:
yield removed
finally:
restore_all_triggers(db)