# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Structured result returned by every migration-service operation.
Every :class:`~mafw.db.migration.service.MigrationManager` operation returns a
single :class:`MigrationOutcome` whose ``result`` field
equals exactly one member of :class:`MigrationResult`. The two front ends (the
``mafw db`` CLI and the :class:`~mafw.processor_library.db_init.SchemaMigrator`
processor) map this one structure to their own conventions, which guarantees
that both entry points agree for identical inputs.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from mafw.db.migration.diff import SchemaDiff
from mafw.db.migration.errors import MigrationError
from mafw.db.migration.history import MigrationHistoryEntry
from mafw.db.migration.script import MigrationScript
[docs]
class MigrationResult(StrEnum):
"""
The mutually exclusive result of a migration-service operation.
Exactly one member is set per :class:`MigrationOutcome`.
.. versionadded:: 2.4
"""
NO_CHANGE = 'no_change'
"""A diff was computed successfully and contains zero differences."""
DIFFERENCE_DETECTED = 'difference_detected'
"""At least one difference was detected (read-only operations)."""
GENERATED = 'generated'
"""Exactly one migration script was generated."""
PENDING = 'pending'
"""One or more pending migrations exist (not yet applied)."""
APPLIED = 'applied'
"""One or more migration scripts were applied successfully."""
UNSUPPORTED_OR_AMBIGUOUS = 'unsupported_or_ambiguous'
"""The diff contains only unsupported or ambiguous differences."""
FAILURE = 'failure'
"""The operation failed; see the outcome's ``error`` field."""
[docs]
@dataclass(frozen=True)
class MigrationOutcome:
"""
The structured result of a single migration-service operation.
:param result: The single mutually exclusive result value.
:type result: MigrationResult
:param diff: The reported schema diff, when applicable.
:type diff: SchemaDiff, Optional
:param scripts: The scripts generated or otherwise relevant to the operation.
:type scripts: tuple[MigrationScript, ...]
:param applied: The history entries recorded as applied, in chronological order.
:type applied: tuple[MigrationHistoryEntry, ...]
:param pending: The pending scripts, in migration order.
:type pending: tuple[MigrationScript, ...]
:param messages: Human-readable report lines.
:type messages: tuple[str, ...]
:param error: The error, populated if and only if ``result`` is ``FAILURE``.
:type error: MigrationError, Optional
:param scoped_equivalence_only: ``True`` when an empty diff must not be read
as full schema equivalence but only as synchronization scoped to the
documented supported-difference types.
:type scoped_equivalence_only: bool
:param legacy: ``True`` when the target database has no migration history.
:type legacy: bool
.. versionadded:: 2.4
"""
result: MigrationResult
diff: SchemaDiff | None = None
scripts: tuple[MigrationScript, ...] = ()
applied: tuple[MigrationHistoryEntry, ...] = ()
pending: tuple[MigrationScript, ...] = ()
messages: tuple[str, ...] = ()
error: MigrationError | None = None
scoped_equivalence_only: bool = False
legacy: bool = False
def __post_init__(self) -> None:
"""Enforce the invariant that ``error`` is set iff ``result`` is ``FAILURE``."""
is_failure = self.result == MigrationResult.FAILURE
if is_failure and self.error is None:
raise ValueError('A FAILURE outcome must carry an error.')
if not is_failure and self.error is not None:
raise ValueError('A non-FAILURE outcome must not carry an error.')
@property
def is_failure(self) -> bool:
"""Whether this outcome represents a failure."""
return self.result == MigrationResult.FAILURE
[docs]
def make_outcome(
result: MigrationResult,
*,
diff: SchemaDiff | None = None,
scripts: Iterable[MigrationScript] = (),
applied: Iterable[MigrationHistoryEntry] = (),
pending: Iterable[MigrationScript] = (),
messages: Iterable[str] = (),
error: MigrationError | None = None,
scoped_equivalence_only: bool = False,
legacy: bool = False,
) -> MigrationOutcome:
"""
Construct a :class:`MigrationOutcome`, normalizing iterables to tuples.
This helper guarantees exactly one ``result`` value and that ``error`` is
populated if and only if ``result`` is :attr:`MigrationResult.FAILURE`.
:param result: The single mutually exclusive result value.
:type result: MigrationResult
:param diff: The reported schema diff, when applicable.
:type diff: SchemaDiff, Optional
:param scripts: The scripts relevant to the operation.
:type scripts: Iterable[MigrationScript]
:param applied: The applied history entries, in chronological order.
:type applied: Iterable[MigrationHistoryEntry]
:param pending: The pending scripts, in migration order.
:type pending: Iterable[MigrationScript]
:param messages: Human-readable report lines.
:type messages: Iterable[str]
:param error: The error, required if and only if ``result`` is ``FAILURE``.
:type error: MigrationError, Optional
:param scoped_equivalence_only: Whether an empty diff is scoped only.
:type scoped_equivalence_only: bool
:param legacy: Whether the database has no migration history.
:type legacy: bool
:return: The constructed outcome.
:rtype: MigrationOutcome
"""
return MigrationOutcome(
result=result,
diff=diff,
scripts=tuple(scripts),
applied=tuple(applied),
pending=tuple(pending),
messages=tuple(messages),
error=error,
scoped_equivalence_only=scoped_equivalence_only,
legacy=legacy,
)
[docs]
def failure(error: MigrationError, *, messages: Sequence[str] = ()) -> MigrationOutcome:
"""
Build a :attr:`MigrationResult.FAILURE` outcome from an error.
:param error: The error identifying the cause of the failure.
:type error: MigrationError
:param messages: Optional human-readable report lines.
:type messages: Sequence[str]
:return: A failure outcome.
:rtype: MigrationOutcome
"""
text = tuple(messages) if messages else (str(error),)
return make_outcome(MigrationResult.FAILURE, error=error, messages=text)