# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Migration-specific exception hierarchy.
All migration errors subclass :class:`~mafw.mafw_errors.MAFwException` so that
callers can catch the whole family with the framework base exception, mirroring
the existing MAFw error style (for example
:class:`~mafw.mafw_errors.InvalidConfigurationError` and
:class:`~mafw.mafw_errors.UnsupportedDatabaseError`).
The migration service converts *expected* conditions (an empty diff, a pending
conflict, an unsupported-only diff) into
:class:`~mafw.db.migration.outcome.MigrationOutcome` values and reserves these
exceptions for genuine faults. Each exception carries the names of the affected
objects/fields so that both the CLI and the processor can emit actionable
diagnostics.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
from collections.abc import Iterable
from mafw.mafw_errors import MAFwException
[docs]
class MigrationError(MAFwException):
"""
Base class for every schema-migration error.
.. versionadded:: 2.4
"""
[docs]
class MigrationConfigurationError(MigrationError):
"""
Raised when the migration configuration is missing or invalid.
Typical causes are a missing or unreadable steering file, a missing database
configuration section, or missing model-module information required for model
discovery.
:param message: The human-readable error message.
:type message: str
:param fields: The names of the invalid configuration fields, when known.
:type fields: Iterable[str], Optional
.. versionadded:: 2.4
"""
def __init__(self, message: str, fields: Iterable[str] | None = None) -> None:
super().__init__(message)
self.fields: tuple[str, ...] = tuple(fields or ())
"""The names of the invalid configuration fields."""
[docs]
class MigrationConflictingTableNameError(MigrationError):
"""
Raised when two or more discovered models declare the same table name.
:param table_name: The shared table name that caused the collision.
:type table_name: str
:param model_names: The names of every model declaring the shared table name.
:type model_names: Iterable[str]
.. versionadded:: 2.4
"""
def __init__(self, table_name: str, model_names: Iterable[str]) -> None:
self.table_name = table_name
"""The shared table name that caused the collision."""
self.model_names: tuple[str, ...] = tuple(model_names)
"""The names of every model declaring the shared table name."""
joined = ', '.join(self.model_names)
super().__init__(f'Conflicting table name {table_name!r} declared by models: {joined}.')
[docs]
class MigrationUnresolvedDirectoryError(MigrationError):
"""
Raised when the effective migration directory cannot be resolved.
This happens when the resolved directory does not exist or is not a readable
directory. The operation is aborted before any script or history record is
created or modified.
:param directory: The directory path that could not be resolved.
:type directory: str
.. versionadded:: 2.4
"""
def __init__(self, directory: str) -> None:
self.directory = directory
"""The directory path that could not be resolved."""
super().__init__(f'The migration directory could not be resolved: {directory!r}.')
[docs]
class MigrationHistoryInconsistencyError(MigrationError):
"""
Raised when the migration history and the directory are inconsistent.
Examples are a history entry with no corresponding script in the directory, a
recorded script whose content changed relative to the directory, or a script
that is out of recorded sequence order.
:param message: The human-readable error message.
:type message: str
:param scripts: The identifiers of the affected scripts, when known.
:type scripts: Iterable[str], Optional
.. versionadded:: 2.4
"""
def __init__(self, message: str, scripts: Iterable[str] | None = None) -> None:
super().__init__(message)
self.scripts: tuple[str, ...] = tuple(scripts or ())
"""The identifiers of the affected scripts."""
[docs]
class MigrationApplicationError(MigrationError):
"""
Raised when the application of a migration script fails.
The affected script is identified by its version so that the operator can
locate and review it. On a transactional backend the schema changes made by
the failing script are rolled back and no history entry is recorded.
:param message: The human-readable error message.
:type message: str
:param version: The version identifier of the failed migration script.
:type version: str, Optional
.. versionadded:: 2.4
"""
def __init__(self, message: str, version: str | None = None) -> None:
super().__init__(message)
self.version = version
"""The version identifier of the failed migration script."""
[docs]
class PartialMigrationError(MigrationApplicationError):
"""
Raised when a migration fails partway on a backend without transactional DDL.
On such a backend (notably MySQL) a mid-script failure cannot be rolled back
transactionally, so the database is left in a partially migrated state. The
error names the last successfully applied statement so the operator can
recover manually.
:param message: The human-readable error message.
:type message: str
:param version: The version identifier of the failed migration script.
:type version: str, Optional
:param last_statement: The last SQL statement that applied successfully.
:type last_statement: str, Optional
.. versionadded:: 2.4
"""
def __init__(self, message: str, version: str | None = None, last_statement: str | None = None) -> None:
super().__init__(message, version=version)
self.last_statement = last_statement
"""The last SQL statement that applied successfully before the failure."""