# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Migration-script representation, naming, rendering and safe writing.
A migration script is a reviewable, editable Python module generated by the
service for one supported schema diff. It declares its version identifier, a
human-readable summary of the detected change, a content checksum, and the
ordered list of :mod:`playhouse.migrate` operations to run inside a
``migrate(migrator, database)`` function.
The file stem is a pure, deterministic function of the version component and a
short stable digest of the *canonicalised* diff: the same diff yields the same
name, and distinct diffs within a directory yield distinct names. Generation
never overwrites an existing file, so a scientist can edit a generated script and
have the edited content applied verbatim.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import hashlib
import re
from dataclasses import dataclass
from pathlib import Path
from mafw.db.migration.diff import DifferenceKind, SchemaDiff
#: Regular expression used to parse the metadata header of a generated script.
_VERSION_RE = re.compile(r"^__migration_version__\s*=\s*['\"](?P<version>[^'\"]+)['\"]", re.MULTILINE)
_CHECKSUM_RE = re.compile(r"^__migration_checksum__\s*=\s*['\"](?P<checksum>[^'\"]+)['\"]", re.MULTILINE)
_REQUIRES_REVIEW_RE = re.compile(r'^__migration_requires_review__\s*=\s*(?P<value>True|False)', re.MULTILINE)
#: Suggested, intentionally non-executing migrator calls for each unsupported
#: difference kind. The scientist must uncomment and complete the operation by
#: hand, because the service never performs a destructive change automatically.
_UNSUPPORTED_HINTS: dict[str, str] = {
DifferenceKind.DROP_COLUMN.value: 'migrator.drop_column({table!r}, {column!r})',
DifferenceKind.DROP_TABLE.value: 'migrator.drop_table({table!r})',
DifferenceKind.ALTER_COLUMN_TYPE.value: 'migrator.alter_column_type({table!r}, {column!r}, <new_field>)',
DifferenceKind.ALTER_NULLABILITY.value: 'migrator.add_not_null({table!r}, {column!r}) # or drop_not_null(...)',
DifferenceKind.ADD_NOT_NULL_COLUMN.value: (
'migrator.add_column({table!r}, {column!r}, <field_with_default>) # backfill data first'
),
DifferenceKind.DROP_INDEX.value: "migrator.drop_index({table!r}, '<index_name>') # index on ({column})",
}
#: Reference list of the operations offered by the per-backend
#: :class:`playhouse.migrate.SchemaMigrator`, embedded verbatim in the docstring
#: of every generated script so a reviewing scientist has the exact signatures to
#: hand while completing a migration. Mirrors the installed Peewee 4.5.3 API and
#: the documentation at
#: https://docs.peewee-orm.com/en/latest/peewee/db_tools.html#module-playhouse.migrate
_MIGRATOR_OPERATIONS: tuple[str, ...] = (
'add_column(table, column_name, field, allow_not_null=False)',
'drop_column(table, column_name, cascade=True)',
'rename_column(table, old_name, new_name)',
'add_not_null(table, column)',
'drop_not_null(table, column)',
'add_column_default(table, column, default)',
'drop_column_default(table, column)',
'alter_column_type(table, column, field, cast=None)',
'rename_table(old_name, new_name)',
'add_index(table, columns, unique=False, using=None, where=None, nulls_distinct=None, name=None)',
'drop_index(table, index_name)',
'add_unique(table, *column_names)',
'add_constraint(table, name, constraint)',
'drop_constraint(table, name)',
'add_foreign_key_constraint(table, column_name, rel, rel_column, ...)',
'drop_foreign_key_constraint(table, column_name)',
'drop_table(table, safe=False, cascade=False, schema=None)',
)
[docs]
@dataclass(frozen=True)
class MigrationScript:
"""
A generated or discovered migration script.
:param version: The sortable version identifier (for example a zero-padded
sequence number).
:type version: str
:param name: The file stem, deterministic from the diff.
:type name: str
:param path: The absolute path of the script on disk.
:type path: Path
:param checksum: The sha-256 checksum of the canonical script content.
:type checksum: str
:param requires_review: Whether the script still carries a pending
review-required flag for one or more unsupported/destructive operations.
A script that requires review is refused by the applier until a scientist
has inspected it and cleared the flag.
:type requires_review: bool
.. versionadded:: 2.4
"""
version: str
name: str
path: Path
checksum: str
requires_review: bool = False
[docs]
def diff_digest(diff: SchemaDiff) -> str:
"""
Return a short stable digest of the canonicalised diff.
:param diff: The schema diff to digest.
:type diff: SchemaDiff
:return: The first twelve hexadecimal characters of the sha-256 digest.
:rtype: str
"""
canonical = diff.canonical_key().encode('utf-8')
return hashlib.sha256(canonical).hexdigest()[:12]
[docs]
def script_name(diff: SchemaDiff, version: str) -> str:
"""
Compute the deterministic, injective file stem for a migration script.
The stem combines the version component with a short stable digest of the
canonicalised diff. ``script_name`` is a pure function: the same diff and
version yield the same name, and distinct diffs yield distinct names.
:param diff: The schema diff the script will carry.
:type diff: SchemaDiff
:param version: The sortable version component.
:type version: str
:return: The file stem (without the ``.py`` extension).
:rtype: str
"""
return f'{version}_{diff_digest(diff)}'
[docs]
def content_checksum(body: str) -> str:
"""
Return the sha-256 checksum of a script body, ignoring the checksum line.
The checksum covers the canonical content of the file so that an unrelated
change to the recorded checksum line cannot mask a real content change.
:param body: The full script body.
:type body: str
:return: The sha-256 hexadecimal digest of the canonical content.
:rtype: str
"""
lines = [line for line in body.splitlines() if not line.startswith('__migration_checksum__')]
canonical = '\n'.join(lines).encode('utf-8')
return hashlib.sha256(canonical).hexdigest()
[docs]
def _operation_lines(diff: SchemaDiff) -> list[str]:
"""Render the ordered operation statements for the supported differences.
Only the *supported* (safe, additive) differences are emitted as executable
operations. Unsupported/destructive differences are rendered separately as
commented, non-executing stubs by :func:`_review_lines`.
:param diff: The schema diff whose supported differences are rendered.
:type diff: SchemaDiff
:return: The list of Python source lines inside the ``migrate`` function.
:rtype: list[str]
"""
lines: list[str] = []
for difference in diff.supported:
if difference.kind == DifferenceKind.ADD_TABLE:
lines.append(f' # Create the new table {difference.table!r}.')
lines.append(f' create_table(database, {difference.table!r})')
elif difference.kind == DifferenceKind.ADD_COLUMN:
column = difference.detail['column']
field_type = difference.detail.get('type', 'TEXT')
lines.append(f' # Add the nullable column {difference.table}.{column}.')
lines.append(
f' operations.append('
f'migrator.add_column({difference.table!r}, {column!r}, '
f'nullable_field({field_type!r})))'
)
elif difference.kind == DifferenceKind.ADD_INDEX:
columns = difference.detail.get('columns', [])
unique = bool(difference.detail.get('unique', False))
lines.append(f' # Add an index on {difference.table}({", ".join(columns)}).')
lines.append(
f' operations.append(migrator.add_index({difference.table!r}, {tuple(columns)!r}, unique={unique}))'
)
if not lines:
lines.append(' # No supported operations were generated for this diff.')
return lines
[docs]
def _review_lines(diff: SchemaDiff) -> list[str]:
"""Render commented stubs for the unsupported/destructive differences.
Each unsupported difference becomes a loud, non-executing comment block that
states the detected change, the data-loss risk, and a suggested (commented)
:mod:`playhouse.migrate` call. The scientist must complete and uncomment the
operation by hand, then clear ``__migration_requires_review__`` to apply it.
:param diff: The schema diff whose unsupported differences are rendered.
:type diff: SchemaDiff
:return: The list of Python source lines inside the review section.
:rtype: list[str]
"""
lines: list[str] = []
for difference in diff.unsupported:
column = difference.detail.get('column', '')
hint = _UNSUPPORTED_HINTS.get(difference.kind.value, '# describe the operation manually')
suggestion = hint.format(table=difference.table, column=column)
lines.append(f' # !! REVIEW REQUIRED: {difference.describe()}')
if difference.reason:
lines.append(f' # Reason: {difference.reason}')
lines.append(' # This change is potentially destructive and may cause DATA LOSS.')
lines.append(' # Complete and uncomment the operation below, then set')
lines.append(' # __migration_requires_review__ = False to allow it to be applied.')
lines.append(f' # operations.append({suggestion})')
return lines
[docs]
def render_script(diff: SchemaDiff, version: str) -> str:
"""
Render an editable migration-script module for a diff.
The generated module declares its version and checksum, documents the change,
and exposes a ``migrate(migrator, database)`` function. Supported (safe,
additive) differences are emitted as executable operations. If the diff
contains any unsupported/destructive difference, the module additionally
declares ``__migration_requires_review__ = True`` and renders a loud,
commented review section: the service refuses to apply such a script until a
scientist has reviewed it, completed the operations by hand, and cleared the
flag. The body is a template a scientist may freely edit before applying.
:param diff: The schema diff to render.
:type diff: SchemaDiff
:param version: The sortable version component.
:type version: str
:return: The full script body as text.
:rtype: str
"""
requires_review = bool(diff.unsupported)
summary = [f'# - {d.describe()}' for d in diff.supported] or ['# - (no supported differences)']
doc_summary = [f'* {d.describe()}' for d in diff.supported] or ['* (no supported differences)']
operation_lines = _operation_lines(diff)
review_lines = _review_lines(diff)
doc_body = ['Detected supported differences:', *doc_summary]
if requires_review:
doc_body += [
'',
'.. warning::',
'',
' This migration also contains UNSUPPORTED / potentially DESTRUCTIVE',
' changes that the service cannot generate automatically. They are',
' listed as commented stubs in the migrate() function below. Review',
' them carefully (they may cause DATA LOSS), complete the operations',
' by hand, and set __migration_requires_review__ = False before',
' applying this migration with "mafw db migrate up".',
'',
'Detected unsupported differences requiring manual review:',
*[f'* {d.describe()}' for d in diff.unsupported],
]
# Embed the available migrator operations so the scientist has the exact
# signatures at hand while reviewing or completing the migration.
doc_body += [
'',
'Available ``migrator`` operations (playhouse.migrate.SchemaMigrator):',
*[f'* ``{op}``' for op in _MIGRATOR_OPERATIONS],
'',
'See https://docs.peewee-orm.com/en/latest/peewee/db_tools.html#schema-migrations',
'for the full description of each operation.',
]
runtime_imports = ['create_table', 'nullable_field']
if requires_review:
# Destructive migrations may need to drop MAFw triggers around a
# table-rebuild operation; expose the context manager for the scientist.
runtime_imports += ['temporary_trigger_removal']
import_block = (
'from mafw.db.migration.runtime import (\n' + ''.join(f' {name},\n' for name in runtime_imports) + ')'
)
review_section = (
['', ' # --- Manual review required (see the module warning above) ---', *review_lines]
if review_lines
else []
)
# For destructive migrations on ALTER-limited backends (notably SQLite) a
# table rebuild cannot drop a table while a trigger points at it. Offer a
# ready-to-enable, commented trigger-bracketing block so the scientist can
# drop MAFw triggers around the operations and recreate them afterwards
# (safe: this mirrors TriggerRefresher).
if requires_review:
apply_block = [
' # On SQLite a column drop/rename/type change rebuilds the table,',
' # which fails if a trigger still points at it. Wrap the operations',
' # in temporary_trigger_removal(database) to drop MAFw triggers for',
' # the duration of the migration and recreate them afterwards (safe;',
' # the triggers are restored even if the migration raises).',
' #',
' # with temporary_trigger_removal(database):',
' # if operations:',
' # migrator.migrate(*operations)',
' if operations:',
' migrator.migrate(*operations)',
]
else:
apply_block = [
' if operations:',
' migrator.migrate(*operations)',
]
header = [
'# Generated by the MAFw schema-migration service.',
'# Review and edit this file before applying it. Your edits are',
'# applied verbatim; the service never regenerates or overwrites it.',
'"""Auto-generated MAFw migration script.',
'',
*doc_body,
'"""',
'from __future__ import annotations',
'',
'import peewee',
'from playhouse.migrate import SchemaMigrator',
'',
import_block,
'',
f"__migration_version__ = '{version}'",
"__migration_checksum__ = '{checksum}'",
f'__migration_requires_review__ = {requires_review}',
'',
'# Human-readable summary of the detected change:',
*summary,
'',
'',
'def migrate(migrator: SchemaMigrator, database: peewee.Database) -> None:',
' """Apply this migration using the supplied per-backend migrator."""',
' operations = []',
*operation_lines,
*review_section,
'',
*apply_block,
'',
]
body_without_checksum = '\n'.join(line for line in header if '{checksum}' not in line)
checksum = content_checksum(body_without_checksum)
return '\n'.join(header).format(checksum=checksum)
[docs]
def write_script(directory: Path, name: str, body: str) -> MigrationScript:
"""
Write a migration script, refusing to overwrite an existing file.
If a file with the target name already exists, its bytes are preserved
unchanged and the existing script is returned (parsed from disk). This is
what lets a scientist edit a generated script and keep the edited content.
:param directory: The resolved migration directory.
:type directory: Path
:param name: The file stem (without extension).
:type name: str
:param body: The script body to write when the file does not yet exist.
:type body: str
:return: The written or pre-existing migration script descriptor.
:rtype: MigrationScript
"""
path = directory / f'{name}.py'
if path.exists():
# Preserve the existing file exactly; return its parsed descriptor.
return load_script(path)
path.write_text(body, encoding='utf-8')
return load_script(path)
[docs]
def load_script(path: Path) -> MigrationScript:
"""
Parse a migration script descriptor from a file on disk.
:param path: The path to the migration script.
:type path: Path
:return: The parsed descriptor.
:rtype: MigrationScript
:raises ValueError: if the file lacks a parseable version header.
"""
text = path.read_text(encoding='utf-8')
version_match = _VERSION_RE.search(text)
if version_match is None:
raise ValueError(f'Migration script {path} has no __migration_version__ declaration.')
version = version_match.group('version')
checksum = content_checksum(text)
review_match = _REQUIRES_REVIEW_RE.search(text)
requires_review = review_match is not None and review_match.group('value') == 'True'
return MigrationScript(
version=version,
name=path.stem,
path=path,
checksum=checksum,
requires_review=requires_review,
)
[docs]
def recorded_checksum(path: Path) -> str | None:
"""
Return the checksum recorded in a script header, if present.
:param path: The path to the migration script.
:type path: Path
:return: The recorded checksum, or ``None`` when absent.
:rtype: str | None
"""
text = path.read_text(encoding='utf-8')
match = _CHECKSUM_RE.search(text)
return match.group('checksum') if match else None