# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Schema differ comparing discovered models with the live database.
Peewee can introspect a live schema (:class:`playhouse.reflection.Introspector`)
and apply individual migration operations (:mod:`playhouse.migrate`), but it has
no built-in schema *diff*. This module is the MAFw-authored differ: it compares
the discovered :class:`~mafw.db.db_model.MAFwBaseModel` subclasses against the
introspected live schema and classifies each detected change as a
*supported* or an *unsupported* difference.
The initial supported set is intentionally conservative and additive: new tables
and new nullable columns. Everything else the differ detects — renames,
drops and other destructive operations, type changes, constraint changes, and
required data backfills — is classified as unsupported. Differences the differ
cannot even detect are documented as undetectable, which is why an empty diff is
only ever reported as *scoped* synchronization and never as full schema
equivalence.
Introspection here is strictly read-only, which is what makes the ``diff`` and
``generate`` service operations non-mutating.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any, Protocol, cast
import peewee
from playhouse.reflection import Introspector
from mafw.db.db_model import MAFwBaseModel
from mafw.db.db_types import PeeweeModelWithMeta
[docs]
class _PeeweeIndex(Protocol):
"""Describe index attributes used by the differ but omitted from Peewee stubs."""
_expressions: Sequence[object]
_where: object | None
_unique: bool
[docs]
class DifferenceKind(StrEnum):
"""
The kind of schema difference detected by the differ.
.. versionadded:: 2.4
"""
ADD_TABLE = 'add_table'
"""A model has no counterpart table in the live schema (supported)."""
ADD_COLUMN = 'add_column'
"""A model declares a nullable column absent from the live table (supported)."""
ADD_NOT_NULL_COLUMN = 'add_not_null_column'
"""A model declares a non-nullable column absent from the live table (unsupported)."""
DROP_COLUMN = 'drop_column'
"""The live table has a column absent from the model (unsupported)."""
DROP_TABLE = 'drop_table'
"""The live schema has a table absent from the models (unsupported)."""
ALTER_COLUMN_TYPE = 'alter_column_type'
"""A column type differs between the model and the live table (unsupported)."""
ALTER_NULLABILITY = 'alter_nullability'
"""A column nullability differs between the model and the live table (unsupported)."""
ADD_INDEX = 'add_index'
"""A model declares an index absent from the live table (supported)."""
DROP_INDEX = 'drop_index'
"""The live table has an index absent from the model (unsupported)."""
[docs]
class SupportLevel(StrEnum):
"""
Whether a detected difference can be safely expressed as a migration.
.. versionadded:: 2.4
"""
SUPPORTED = 'supported'
"""The differ can both detect and safely express the difference."""
UNSUPPORTED = 'unsupported'
"""The difference is detected but cannot be safely expressed automatically."""
#: The additive-only set of difference kinds classified as supported (OQ-1).
SUPPORTED_KINDS: frozenset[DifferenceKind] = frozenset(
{
DifferenceKind.ADD_TABLE,
DifferenceKind.ADD_COLUMN,
DifferenceKind.ADD_INDEX,
}
)
"""The additive-only supported difference kinds: new tables, new nullable columns, new indexes."""
[docs]
@dataclass(frozen=True)
class Difference:
"""
One detected schema difference.
:param kind: The kind of difference.
:type kind: DifferenceKind
:param table: The table the difference affects.
:type table: str
:param support: Whether the difference is supported or unsupported.
:type support: SupportLevel
:param detail: Extra structured information (column name, type, nullability).
:type detail: Mapping[str, Any]
:param reason: Why the difference is unsupported, for the report.
:type reason: str, Optional
.. versionadded:: 2.4
"""
kind: DifferenceKind
table: str
support: SupportLevel
detail: Mapping[str, Any] = field(default_factory=dict)
reason: str | None = None
[docs]
def describe(self) -> str:
"""Return a short human-readable description of the difference.
:return: A one-line description suitable for a report or script comment.
:rtype: str
"""
column = self.detail.get('column')
target = f'{self.table}.{column}' if column else self.table
text = f'{self.kind.value} on {target}'
if self.reason:
text += f' ({self.reason})'
return text
[docs]
@dataclass(frozen=True)
class SchemaDiff:
"""
The structured comparison between the models and the live schema.
The supported and unsupported lists partition the detected differences: each
detected difference appears in exactly one list.
:param supported: The supported differences, in deterministic order.
:type supported: tuple[Difference, ...]
:param unsupported: The unsupported differences, in deterministic order.
:type unsupported: tuple[Difference, ...]
.. versionadded:: 2.4
"""
supported: tuple[Difference, ...] = ()
unsupported: tuple[Difference, ...] = ()
@property
def is_empty(self) -> bool:
"""Whether the diff contains no supported and no unsupported difference."""
return not self.supported and not self.unsupported
@property
def has_supported(self) -> bool:
"""Whether the diff contains at least one supported difference."""
return bool(self.supported)
@property
def all_differences(self) -> tuple[Difference, ...]:
"""The supported and unsupported differences concatenated."""
return self.supported + self.unsupported
[docs]
def canonical_key(self) -> str:
"""Return a stable string key uniquely identifying the diff content.
Two diffs that describe the same changes produce the same key regardless
of discovery order. This underpins deterministic script naming.
:return: A canonical, order-independent key for the whole diff.
:rtype: str
"""
parts = sorted(
f'{d.kind.value}:{d.table}:{d.detail.get("column", "")}:{d.detail.get("type", "")}'
for d in self.all_differences
)
return '\n'.join(parts)
[docs]
def _diff_sort_key(difference: Difference) -> tuple[str, str, str]:
"""Sort key giving a deterministic, stable ordering of differences.
:param difference: The difference to produce a sort key for.
:type difference: Difference
:return: A tuple used to order differences deterministically.
:rtype: tuple[str, str, str]
"""
return (difference.table, difference.kind.value, str(difference.detail.get('column', '')))
[docs]
def _model_columns(model: type[MAFwBaseModel]) -> dict[str, peewee.Field]:
"""Return the mapping of column name to field for a model.
:param model: The model whose columns are requested.
:type model: type[MAFwBaseModel]
:return: Mapping of database column name to the peewee field.
:rtype: dict[str, peewee.Field]
"""
meta = cast(PeeweeModelWithMeta, model)._meta
return {field_obj.column_name: field_obj for field_obj in meta.sorted_fields}
[docs]
def _field_type_token(field_type: Any) -> str:
"""Normalise a peewee ``field_type`` token for comparison.
The token is upper-cased and its trailing length/precision specifier (for
example the ``(255)`` in ``VARCHAR(255)``) is dropped, so that equivalent
types compare equal across backends and reflection quirks.
:param field_type: The ``field_type`` token from a model field or a
reflected column's ``field_class``.
:type field_type: Any
:return: The normalised, comparable type token, or ``''`` when the input is
not a plain string token (for example a descriptor leaking through).
:rtype: str
"""
if not isinstance(field_type, str):
return ''
token = field_type.upper()
return token.split('(', 1)[0].strip()
#: Maps normalised type tokens to a coarse "family". Two columns are only
#: reported as a type change when they fall into *different* families, so benign
#: within-family differences — notably the SQLite ``TEXT`` / ``VARCHAR`` / ``CHAR``
#: equivalence, or ``INT`` vs ``BIGINT`` — never raise a false positive. A token
#: that is not mapped forms its own singleton family and so compares only to
#: itself.
_TYPE_FAMILIES: dict[str, str] = {
'TEXT': 'string',
'VARCHAR': 'string',
'CHAR': 'string',
'INT': 'integer',
'INTEGER': 'integer',
'BIGINT': 'integer',
'SMALLINT': 'integer',
'BOOL': 'integer',
'FLOAT': 'real',
'REAL': 'real',
'DOUBLE': 'real',
'DECIMAL': 'real',
}
[docs]
def _type_family(token: str) -> str:
"""Return the coarse type family for a normalised type token.
:param token: A normalised type token from :func:`_field_type_token`.
:type token: str
:return: The family name, or the token itself when it has no mapped family.
:rtype: str
"""
return _TYPE_FAMILIES.get(token, token)
[docs]
def _detect_type_change(
table: str,
column_name: str,
field_obj: peewee.Field,
live_column: Any,
) -> Difference | None:
"""Return an ``ALTER_COLUMN_TYPE`` difference when a column type changed.
The comparison uses peewee's own ``field_type`` token on both sides — the
model field's token versus the token of the ``field_class`` that
:class:`playhouse.reflection.Introspector` inferred for the live column —
reduced to a coarse *type family* so that benign within-family differences
(for example the SQLite ``TEXT`` / ``VARCHAR`` equivalence, or ``INT`` vs
``BIGINT``) do not cause false positives. Primary-key columns are skipped
because reflection reports them as ``AUTO``/``INT`` inconsistently across
backends. A detected change is classified as *unsupported*: altering a column
type can be lossy and must be reviewed.
:param table: The table the column belongs to.
:type table: str
:param column_name: The database column name.
:type column_name: str
:param field_obj: The model field.
:type field_obj: peewee.Field
:param live_column: The reflected live column metadata.
:type live_column: Any
:return: An unsupported type-change difference, or ``None`` when unchanged.
:rtype: Difference | None
"""
if getattr(field_obj, 'primary_key', False) or getattr(live_column, 'primary_key', False):
return None
# Skip foreign-key columns. A FK column's type is governed by the referenced
# key, and reflection reports a FK column's ``field_class`` as
# ``ForeignKeyField`` whose ``field_type`` is an instance *property*
# descriptor (not a type token) when accessed on the class — so a naive
# comparison would mis-read it as ``<property object ...>`` and flag every FK
# column as a type change. Genuine FK relationship changes are a constraint
# concern handled (or deliberately left for manual review) separately.
if isinstance(field_obj, peewee.ForeignKeyField):
return None
live_field_class = getattr(live_column, 'field_class', None)
if live_field_class is None:
return None
if isinstance(live_field_class, type) and issubclass(live_field_class, peewee.ForeignKeyField):
return None
model_token = _field_type_token(field_obj.field_type)
live_token = _field_type_token(getattr(live_field_class, 'field_type', None))
# A non-token (e.g. a property descriptor leaking through) is not comparable;
# never raise a spurious type change from it.
if not model_token or not live_token:
return None
if _type_family(model_token) == _type_family(live_token):
return None
return Difference(
kind=DifferenceKind.ALTER_COLUMN_TYPE,
table=table,
support=SupportLevel.UNSUPPORTED,
detail={'column': column_name, 'type': model_token, 'current_type': live_token},
reason=f'column type differs: database has {live_token}, model declares {model_token}',
)
[docs]
def _model_index_signatures(model: type[MAFwBaseModel]) -> set[tuple[tuple[str, ...], bool]]:
"""Return the set of ``(columns, unique)`` signatures declared by a model.
Only plain column indexes are considered; partial and expression indexes
carry no column signature and are left to hand-authored migrations, matching
the behaviour of :func:`playhouse.schema_diff.diff_models`.
:param model: The model whose declared indexes are inspected.
:type model: type[MAFwBaseModel]
:return: The set of ``(ordered column names, unique flag)`` signatures.
:rtype: set[tuple[tuple[str, ...], bool]]
"""
signatures: set[tuple[tuple[str, ...], bool]] = set()
meta = cast(PeeweeModelWithMeta, model)._meta
for index in meta.fields_to_index():
if not isinstance(index, peewee.Index):
continue # Raw SQL index declarations are out of scope.
# Peewee's runtime Index has these fields, though its stubs omit them.
typed_index = cast(_PeeweeIndex, index)
parts = typed_index._expressions
if typed_index._where is None and all(isinstance(part, peewee.Field) for part in parts):
fields = [part for part in parts if isinstance(part, peewee.Field)]
columns = tuple(part.column_name for part in fields)
signatures.add((columns, typed_index._unique))
return signatures
[docs]
def _live_index_signatures(database: peewee.Database, table: str) -> set[tuple[tuple[str, ...], bool]]:
"""Return the set of ``(columns, unique)`` signatures present on a live table.
Primary-key and partial/expression indexes are excluded, mirroring
:func:`playhouse.schema_diff.diff_models`.
:param database: The live database.
:type database: peewee.Database
:param table: The table to inspect.
:type table: str
:return: The set of live ``(ordered column names, unique flag)`` signatures.
:rtype: set[tuple[tuple[str, ...], bool]]
"""
signatures: set[tuple[tuple[str, ...], bool]] = set()
try:
indexes = database.get_indexes(table)
except (peewee.DatabaseError, NotImplementedError): # pragma: no cover - backend specific
return signatures
for index in indexes:
name = index.name
if name == 'PRIMARY' or name.endswith('_pkey') or name.startswith('sqlite_autoindex_'):
continue
if None in index.columns:
continue # Partial / expression index: no comparable signature.
signatures.add((tuple(index.columns), bool(index.unique)))
return signatures
[docs]
def _diff_indexes(
model: type[MAFwBaseModel],
table: str,
database: peewee.Database,
supported: list[Difference],
unsupported: list[Difference],
) -> None:
"""Compare a model's declared indexes against the live table's indexes.
Adding an index is additive and therefore *supported*; dropping an index
(present in the database but not declared by the model) is *unsupported* and
left for review, since an index may have been added deliberately outside the
model.
:param model: The model being compared.
:type model: type[MAFwBaseModel]
:param table: The table name.
:type table: str
:param database: The live database.
:type database: peewee.Database
:param supported: The list that collects supported differences (mutated).
:type supported: list[Difference]
:param unsupported: The list that collects unsupported differences (mutated).
:type unsupported: list[Difference]
"""
model_indexes = _model_index_signatures(model)
live_indexes = _live_index_signatures(database, table)
for columns, unique in sorted(model_indexes - live_indexes):
supported.append(
Difference(
kind=DifferenceKind.ADD_INDEX,
table=table,
support=SupportLevel.SUPPORTED,
detail={'columns': list(columns), 'unique': unique, 'column': ','.join(columns)},
)
)
for columns, unique in sorted(live_indexes - model_indexes):
unsupported.append(
Difference(
kind=DifferenceKind.DROP_INDEX,
table=table,
support=SupportLevel.UNSUPPORTED,
detail={'columns': list(columns), 'unique': unique, 'column': ','.join(columns)},
reason='index exists in the database but not in the model (possible drop)',
)
)
[docs]
def compute_diff(
models: Sequence[type[MAFwBaseModel]],
database: peewee.Database,
*,
ignore_tables: frozenset[str] = frozenset(),
) -> SchemaDiff:
"""
Compute the schema diff between the discovered models and the live database.
The live schema is obtained with :class:`playhouse.reflection.Introspector`
in a strictly read-only manner. Detected differences are partitioned into the
supported (additive) and unsupported lists.
:param models: The discovered models to compare.
:type models: Sequence[type[MAFwBaseModel]]
:param database: The live database to introspect.
:type database: peewee.Database
:param ignore_tables: Live table names to exclude from drop detection (for
example the internal migration-history table).
:type ignore_tables: frozenset[str]
:return: The computed schema diff.
:rtype: SchemaDiff
"""
# The third-party factory is untyped; its return value is an Introspector.
from_database = cast(Callable[[peewee.Database], Introspector], Introspector.from_database)
introspector = from_database(database)
# Peewee's introspection method is also untyped in its published stubs.
metadata = cast(Any, introspector.introspect()) # type: ignore[no-untyped-call]
live_columns: Mapping[str, Mapping[str, Any]] = metadata.columns
live_tables = set(live_columns) - set(ignore_tables)
supported: list[Difference] = []
unsupported: list[Difference] = []
for model in models:
table = cast(PeeweeModelWithMeta, model)._meta.table_name
if table not in live_tables:
# Whole new table: an additive, supported difference.
supported.append(
Difference(
kind=DifferenceKind.ADD_TABLE,
table=table,
support=SupportLevel.SUPPORTED,
detail={'table': table},
)
)
continue
model_cols = _model_columns(model)
live_cols = live_columns[table]
# Columns present in the model but not in the live table.
for column_name, field_obj in model_cols.items():
if column_name in live_cols:
# Column present on both sides: check for a column type change.
type_change = _detect_type_change(table, column_name, field_obj, live_cols[column_name])
if type_change is not None:
unsupported.append(type_change)
continue
nullable = bool(getattr(field_obj, 'null', False))
if nullable:
supported.append(
Difference(
kind=DifferenceKind.ADD_COLUMN,
table=table,
support=SupportLevel.SUPPORTED,
detail={
'column': column_name,
'type': field_obj.field_type,
'nullable': True,
},
)
)
else:
# A new non-nullable column needs a default/backfill and is unsafe.
unsupported.append(
Difference(
kind=DifferenceKind.ADD_NOT_NULL_COLUMN,
table=table,
support=SupportLevel.UNSUPPORTED,
detail={'column': column_name, 'type': field_obj.field_type, 'nullable': False},
reason='new non-nullable column requires a default or data backfill',
)
)
# Columns present in the live table but not in the model (potential drop).
for column_name in live_cols:
if column_name not in model_cols:
unsupported.append(
Difference(
kind=DifferenceKind.DROP_COLUMN,
table=table,
support=SupportLevel.UNSUPPORTED,
detail={'column': column_name},
reason='column exists in the database but not in the model (possible drop or rename)',
)
)
# Indexes: compare the model's declared indexes against the live ones.
_diff_indexes(model, table, database, supported, unsupported)
# Tables present in the live schema but not in the models are intentionally
# NOT reported as drops. A database legitimately contains tables beyond the
# currently-discovered model set — MAFw standard tables, tables owned by
# other plugins, or tables from a different analysis sharing the database —
# and the migration history is per-database irrespective of the active plugin
# set (OQ-4). Treating every unknown live table as a potential drop would
# flood the diff with false, destructive warnings, so whole-table drops are
# left to deliberate, hand-authored migrations rather than detected here.
supported.sort(key=_diff_sort_key)
unsupported.sort(key=_diff_sort_key)
return SchemaDiff(supported=tuple(supported), unsupported=tuple(unsupported))