# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Model discovery for the migration service.
Discovery builds the deduplicated set of :class:`~mafw.db.db_model.MAFwBaseModel`
subclasses to compare against the live database schema. Models are collected
exclusively from the global model register
(:data:`~mafw.db.db_model.mafw_model_register`) so the user never has to
enumerate changed fields or model modules.
The register maps a table name to a single model, so deduplication is intrinsic:
the same model registered twice occupies one key. Discovery additionally
recomputes each model's ``_meta.table_name`` and detects cross-model table-name
collisions (for example two model classes that resolve to the same explicit
``Meta.table_name``); on a collision it raises a
:class:`~mafw.db.migration.errors.MigrationConflictingTableNameError` so that no
diff is produced from an ambiguous model set.
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
from typing import cast
from mafw.db.db_model import MAFwBaseModel, mafw_model_register
from mafw.db.db_types import PeeweeModelWithMeta
from mafw.db.migration.errors import MigrationConflictingTableNameError
from mafw.db.model_register import ModelRegister
[docs]
@dataclass(frozen=True)
class DiscoveryResult:
"""
The result of a model-discovery operation.
:param models: The deduplicated, table-name-unique models to compare.
:type models: tuple[type[MAFwBaseModel], ...]
:param table_names: The set of table names of the discovered models.
:type table_names: frozenset[str]
.. versionadded:: 2.4
"""
models: tuple[type[MAFwBaseModel], ...]
table_names: frozenset[str]
[docs]
def _table_name(model: type[MAFwBaseModel]) -> str:
"""Return the database table name declared by a model.
:param model: The model whose table name is requested.
:type model: type[MAFwBaseModel]
:return: The table name from the model metadata.
:rtype: str
"""
return cast(str, cast(PeeweeModelWithMeta, model)._meta.table_name)
[docs]
def discover(registry: ModelRegister = mafw_model_register) -> DiscoveryResult:
"""
Discover every registered :class:`~mafw.db.db_model.MAFwBaseModel` subclass.
Collection is exclusive to the supplied register (defaulting to the global
:data:`~mafw.db.db_model.mafw_model_register`). Non-:class:`~mafw.db.db_model.MAFwBaseModel`
classes, unregistered classes, and the internal
:class:`~mafw.db.migration.history.MigrationHistory` model are excluded. The
returned set is deduplicated and unique by table name.
:param registry: The model register to discover from. Defaults to the global
register.
:type registry: ModelRegister
:return: The deduplicated discovery result.
:rtype: DiscoveryResult
:raises MigrationConflictingTableNameError: if two or more discovered models
resolve to the same table name.
"""
# Imported lazily to avoid a circular import: history imports db_model which
# is imported here, and history must be excluded from discovery (Req 2.9).
from mafw.db.migration.history import MigrationHistory
# ``items()`` returns a snapshot ``list[(table_name, model)]`` taken under
# the register lock, so iteration is safe even if the register mutates.
seen: set[int] = set()
discovered: list[type[MAFwBaseModel]] = []
by_table: defaultdict[str, list[type[MAFwBaseModel]]] = defaultdict(list)
for _, model in registry.items():
# Keep only concrete MAFwBaseModel subclasses; exclude the history model.
if not (isinstance(model, type) and issubclass(model, MAFwBaseModel)):
continue
if model is MAFwBaseModel or issubclass(model, MigrationHistory):
continue
# Deduplicate the same model object appearing more than once.
if id(model) in seen:
continue
seen.add(id(model))
table = _table_name(model)
by_table[table].append(model)
discovered.append(model)
# Detect cross-model collisions: two distinct model classes with one table.
collisions = {table: models for table, models in by_table.items() if len(models) > 1}
if collisions:
# Report the first collision deterministically (sorted by table name).
table = sorted(collisions)[0]
names = [m.__name__ for m in collisions[table]]
raise MigrationConflictingTableNameError(table, names)
return DiscoveryResult(
models=tuple(discovered),
table_names=frozenset(by_table),
)