Registry¶

def _make_metadata(self) -> sqlalchemy.MetaData:
    """Create a fresh SQLAlchemy metadata container for this registry.

    When the registry is pinned to a default schema, every metadata object
    created here is initialized with that schema so table creation,
    reflection, and migration preparation all operate in the same namespace.

    Returns:
        sqlalchemy.MetaData: New metadata object for the registry.
    """
    if self.db_schema is not None:
        return sqlalchemy.MetaData(schema=self.db_schema)
    return sqlalchemy.MetaData()

def extra_name_check(self, name: Any) -> bool:
    """Validate one entry key from the registry `extra` mapping.

    Saffier accepts arbitrary names for extra databases, but empty or
    non-string keys make later lookups ambiguous. This helper centralizes the
    validation and logs user-facing diagnostics before registry creation
    proceeds.

    Args:
        name: Candidate extra-database name.

    Returns:
        bool: `True` when the key is usable, otherwise `False`.
    """
    if not isinstance(name, str):
        logger.error("Extra database name: %r is not a string.", name)
        return False
    if not name.strip():
        logger.error('Extra database name: "%s" is empty.', name)
        return False
    if name.strip() != name:
        logger.warning(
            'Extra database name: "%s" starts or ends with whitespace characters.', name
        )
    return True

def _get_relation_target_name(self, relation: Any) -> str | None:
    if isinstance(relation, str):
        return relation
    if isinstance(relation, type):
        return relation.__name__
    return None

def _is_auto_through_model(self, model_class: type[Any]) -> bool:
    """Detect auto-generated many-to-many through models.

    Registry copying skips Saffier-generated through models during the first
    pass because they are recreated or patched after their owning models have
    been copied. This helper matches both the older and newer generated
    naming conventions so copied registries preserve many-to-many behavior
    without duplicating synthetic models.

    Args:
        model_class: Model class to inspect.

    Returns:
        bool: `True` if the model looks like an auto-generated through model.
    """
    from saffier.core.db.fields.base import ManyToManyField

    for owner_model in self.models.values():
        for field_name, field in owner_model.fields.items():
            if not isinstance(field, ManyToManyField):
                continue
            through_model = getattr(field, "through", None)
            if through_model is not model_class:
                continue
            target_name = self._get_relation_target_name(field.to) or field.target.__name__
            expected_old_name = f"{owner_model.__name__}{target_name}"
            expected_old_table = f"{owner_model.__name__.lower()}s_{target_name}s".lower()

            expected_new_name = f"{owner_model.__name__}{field_name.capitalize()}Through"
            expected_new_table = expected_new_name.lower()
            if owner_model.meta.table_prefix:
                expected_old_table = f"{owner_model.meta.table_prefix}_{expected_old_table}"
                expected_new_table = f"{owner_model.meta.table_prefix}_{expected_new_table}"

            return (
                model_class.__name__ == expected_old_name
                and getattr(model_class.meta, "tablename", None) == expected_old_table
            ) or (
                model_class.__name__ == expected_new_name
                and getattr(model_class.meta, "tablename", None) == expected_new_table
            )
    return False

def _copy_dependencies(self, model_class: type[Any], skipped: set[str]) -> set[str]:
    from saffier.core.db.fields.base import ForeignKey

    dependencies: set[str] = set()
    for field in model_class.fields.values():
        if isinstance(field, ForeignKey):
            target_name = self._get_relation_target_name(field.to)
            if (
                target_name
                and target_name in self.models
                and target_name != model_class.__name__
            ):
                dependencies.add(target_name)
    dependencies.difference_update(skipped)
    return dependencies

def _sorted_model_names_for_copy(self) -> list[str]:
    """Topologically order model names for registry copying.

    Models with foreign-key dependencies should be copied after their
    targets whenever possible so string references and through-model patches
    are minimized. Auto-generated through models are skipped entirely here
    because they are handled separately after the base copy pass.

    Returns:
        list[str]: Model names ordered for safe copying.
    """
    skipped = {
        name
        for name, model_class in self.models.items()
        if self._is_auto_through_model(model_class)
    }
    remaining = {
        name: self._copy_dependencies(model_class, skipped)
        for name, model_class in self.models.items()
        if name not in skipped
    }
    ordered: list[str] = []
    resolved: set[str] = set()

    while remaining:
        ready = [name for name, deps in remaining.items() if deps.issubset(resolved)]
        if not ready:
            ordered.extend(remaining.keys())
            break
        for name in ready:
            ordered.append(name)
            resolved.add(name)
            remaining.pop(name)
    return ordered

def _copy_model_to_registry(
    self,
    model_class: type[Any],
    registry: "Registry",
    *,
    pending_m2m_patches: list[tuple[str, str, str]],
) -> type[Any]:
    """Clone one model class into a target registry.

    The copied class receives copied field and manager instances so later
    mutations in migration preparation or isolated test registries do not
    leak back into the source registry. Same-registry relation targets are
    rewritten to string references until the target registry has all of its
    models in place, at which point pending many-to-many through models are
    patched separately.

    Args:
        model_class: Source model to copy.
        registry: Target registry receiving the copied model.
        pending_m2m_patches: Collector for many-to-many fields whose through
            model must be rebound after all models have been copied.

    Returns:
        type[Any]: Newly created copied model class.
    """
    from saffier.core.db.fields.base import ForeignKey, ManyToManyField
    from saffier.core.db.models.managers import Manager
    from saffier.core.utils.models import create_saffier_model

    definitions: dict[str, Any] = {}
    manager_annotations: dict[str, Any] = {}
    existing_annotations = dict(getattr(model_class, "__annotations__", {}))
    for field_name, field in model_class.fields.items():
        field_copy = copy.copy(field)
        if hasattr(field_copy, "_target"):
            delattr(field_copy, "_target")
        if isinstance(field_copy, ForeignKey):
            target_name = self._get_relation_target_name(field_copy.to)
            if target_name and target_name in self.models:
                field_copy.to = target_name
        elif isinstance(field_copy, ManyToManyField):
            target_name = self._get_relation_target_name(field_copy.to)
            if target_name and target_name in self.models:
                field_copy.to = target_name

            through_name = self._get_relation_target_name(field_copy.through)
            if through_name and through_name in self.models:
                if self._is_auto_through_model(self.models[through_name]):
                    field_copy.through = None
                elif through_name in registry.models:
                    field_copy.through = registry.models[through_name]
                else:
                    pending_m2m_patches.append(
                        (model_class.__name__, field_name, through_name)
                    )

        definitions[field_name] = field_copy

    for manager_name in getattr(model_class.meta, "managers", []):
        manager = getattr(model_class, manager_name, None)
        if isinstance(manager, Manager):
            definitions[manager_name] = copy.copy(manager)
            manager_annotations[manager_name] = existing_annotations.get(
                manager_name,
                ClassVar[Any],
            )

    if manager_annotations:
        definitions["__annotations__"] = {
            **existing_annotations,
            **manager_annotations,
        }

    meta = type(
        "Meta",
        (),
        {
            "registry": registry,
            "tablename": getattr(model_class.meta, "tablename", None),
            "table_prefix": getattr(model_class.meta, "table_prefix", None),
            "unique_together": list(getattr(model_class.meta, "unique_together", []) or []),
            "indexes": list(getattr(model_class.meta, "indexes", []) or []),
            "constraints": list(getattr(model_class.meta, "constraints", []) or []),
            "reflect": getattr(model_class.meta, "reflect", False),
            "abstract": getattr(model_class.meta, "abstract", False),
            "model_engine": getattr(model_class.meta, "model_engine", None),
        },
    )

    return create_saffier_model(
        model_class.__name__,
        model_class.__module__,
        __definitions__=definitions,
        __metadata__=meta,
        __qualname__=model_class.__qualname__,
        __bases__=model_class.__bases__,
    )

def __copy__(self) -> "Registry":
    """Copy the registry and all registered models into an isolated registry instance.

    The copy operation is intentionally deep for model definitions: each
    copied model gets copied fields, managers, metadata, and many-to-many
    relation descriptors so migration preparation or test isolation does not
    mutate the live application registry.

    Returns:
        Registry: A new registry instance containing copied model classes.
    """
    registry_copy = type(self)(
        self.database,
        schema=self.db_schema,
        extra=self.extra,
        automigrate_config=self._automigrate_config,
        model_engine=self.model_engine,
    )
    pending_m2m_patches: list[tuple[str, str, str]] = []

    for model_name in self._sorted_model_names_for_copy():
        self._copy_model_to_registry(
            self.models[model_name],
            registry_copy,
            pending_m2m_patches=pending_m2m_patches,
        )

    for model_name, model_class in self.reflected.items():
        if model_name in registry_copy.models or model_name in registry_copy.reflected:
            continue
        self._copy_model_to_registry(
            model_class,
            registry_copy,
            pending_m2m_patches=pending_m2m_patches,
        )

    for owner_name, field_name, through_name in pending_m2m_patches:
        from saffier.core.db.fields.base import ManyToManyField
        from saffier.core.db.relationships.relation import Relation

        through_model = registry_copy.models.get(through_name) or registry_copy.reflected.get(
            through_name
        )
        if through_model is None:
            continue
        owner_model = registry_copy.models[owner_name]
        field = cast("ManyToManyField", owner_model.fields[field_name])
        field.through = through_model
        setattr(
            owner_model,
            settings.many_to_many_relation.format(key=field_name),
            Relation(through=through_model, to=field.target, owner=owner_model),
        )

    registry_copy.pattern_models = dict(self.pattern_models)
    if hasattr(self, "tenant_models") and hasattr(registry_copy, "tenant_models"):
        registry_copy.tenant_models = {
            name: model
            for name in self.tenant_models
            if (model := registry_copy.models.get(name) or registry_copy.reflected.get(name))
            is not None
        }
    registry_copy._pattern_reflected_dbs = set(self._pattern_reflected_dbs)
    registry_copy._content_type_models_bound = set(self._content_type_models_bound)
    if self.content_type is not None:
        with_content_type = registry_copy.models.get(
            "ContentType"
        ) or registry_copy.reflected.get("ContentType")
        if with_content_type is None:
            with_content_type = self.content_type
        registry_copy.content_type = with_content_type
        registry_copy._attach_content_type_to_registered_models()
    return registry_copy

def _set_content_type(self, with_content_type: bool | type[Any]) -> None:
    """Enable and normalize content-type support for the registry.

    The registry accepts either `True` to use the built-in content type
    model or an explicit model class. Abstract or detached content type
    models are copied into this registry as needed, then all registered
    concrete models are updated so they expose a content-type relation and
    pre-save hook.

    Args:
        with_content_type: Flag or model class describing the content-type
            model to use.

    Raises:
        TypeError: If the provided value is neither a boolean flag nor a
            model class.
    """
    from saffier.contrib.contenttypes.models import ContentType

    content_type_model = ContentType if with_content_type is True else with_content_type
    if not isinstance(content_type_model, type):
        raise TypeError("with_content_type must be True/False or a model type.")

    if getattr(content_type_model.meta, "abstract", False):
        meta = type(
            "Meta",
            (),
            {
                "registry": self,
                "tablename": "contenttypes",
            },
        )
        content_type_model = type("ContentType", (content_type_model,), {"Meta": meta})
    elif getattr(content_type_model.meta, "registry", None) in (None, False):
        if not getattr(content_type_model.meta, "tablename", None):
            content_type_model.meta.tablename = "contenttypes"
        content_type_model = content_type_model.add_to_registry(self, name="ContentType")

    registered_content_type = self.models.get("ContentType")
    if registered_content_type is not None:
        content_type_model = registered_content_type

    self.content_type = content_type_model
    self._attach_content_type_to_registered_models()

def _attach_content_type_to_registered_models(self) -> None:
    """Attach content-type integration to every registered concrete model.

    This method is idempotent. It is called after registry creation, after
    registry copies, and whenever a registry gains a new content type model.
    """
    if self.content_type is None:
        return
    for model in self.models.values():
        self._attach_content_type_to_model(model)

def _handle_model_registration(self, model_class: type[Any]) -> None:
    if self.content_type is None:
        return
    self._attach_content_type_to_model(model_class)

def _attach_content_type_to_model(self, model_class: type[Any]) -> None:
    """Install or normalize the content-type field on one model class.

    The helper handles three cases: models that already declare a
    `ContentTypeField`, models that declare an alternate content type field
    name, and models that need the default synthetic `content_type` field
    injected. It also refreshes reverse descriptors, deletion behavior, and
    proxy-model caches when the model definition changes.

    Args:
        model_class: Concrete model to update.
    """
    if self.content_type is None:
        return
    if model_class in (self.content_type, getattr(self.content_type, "proxy_model", None)):
        return
    if getattr(model_class.meta, "abstract", False):
        return
    if getattr(model_class, "is_proxy_model", False):
        return
    if model_class.__name__ in self.reflected:
        return

    from saffier.contrib.contenttypes.fields import ContentTypeField
    from saffier.core.db.models.metaclasses import _set_related_name_for_foreign_keys

    if "content_type" in model_class.fields:
        if isinstance(model_class.fields["content_type"], ContentTypeField):
            if getattr(model_class.meta, "is_tenant", False):
                model_class.fields["content_type"].no_constraint = True
            target_registry = getattr(self.content_type.meta, "registry", None)
            if (
                getattr(model_class.meta, "registry", None) is not target_registry
                or getattr(model_class, "database", None)
                is not getattr(self.content_type, "database", None)
                or getattr(model_class.meta, "is_tenant", False)
            ):
                self.content_type.__require_model_based_deletion__ = True
            self._bind_content_type_pre_save(model_class)
        return

    has_content_type_field = any(
        isinstance(field, ContentTypeField) for field in model_class.fields.values()
    )
    if has_content_type_field:
        content_type_fields = {
            field_name: field
            for field_name, field in model_class.fields.items()
            if isinstance(field, ContentTypeField)
        }
        for field_name, field in content_type_fields.items():
            field.owner = model_class
            field.registry = self
            if getattr(model_class.meta, "is_tenant", False):
                field.no_constraint = True
            if isinstance(field.to, str) and field.to == "ContentType":
                field.to = self.content_type
                if hasattr(field, "_target"):
                    delattr(field, "_target")
            auto_related_names = {
                model_class.__name__.lower(),
                f"{model_class.__name__.lower()}s_set",
            }
            desired_related_name = (
                f"reverse_{model_class.__name__.lower()}"
                if field.related_name in auto_related_names or field.related_name is None
                else field.related_name
            )

            if desired_related_name not in model_class.meta.related_names:
                if field.related_name in model_class.meta.related_names:
                    previous_related_name = cast("str", field.related_name)
                    model_class.meta.related_names.discard(previous_related_name)
                    model_class.meta.related_fields.pop(previous_related_name, None)
                    model_class.meta.related_names_mapping.pop(previous_related_name, None)

                    target_meta = field.target.meta
                    target_meta.related_fields.pop(previous_related_name, None)
                    target_meta.related_names_mapping.pop(previous_related_name, None)
                    target_meta.fields.pop(previous_related_name, None)
                    target_meta.fields_mapping.pop(previous_related_name, None)
                    if hasattr(field.target, previous_related_name):
                        delattr(field.target, previous_related_name)
                    proxy_target = getattr(field.target, "proxy_model", None)
                    if proxy_target is not None and hasattr(
                        proxy_target, previous_related_name
                    ):
                        delattr(proxy_target, previous_related_name)

                field.related_name = desired_related_name
                related_names = _set_related_name_for_foreign_keys(
                    {field_name: field},
                    cast(Any, model_class),
                )
                model_class.meta.related_names.update(related_names)
            target_registry = getattr(self.content_type.meta, "registry", None)
            if (
                getattr(model_class.meta, "registry", None) is not target_registry
                or getattr(model_class, "database", None)
                is not getattr(self.content_type, "database", None)
                or getattr(model_class.meta, "is_tenant", False)
            ):
                self.content_type.__require_model_based_deletion__ = True
        self._bind_content_type_pre_save(model_class)
        return

    related_name = f"reverse_{model_class.__name__.lower()}"
    if hasattr(self.content_type, related_name):
        raise RuntimeError(
            f"Duplicate related content type name generated: {related_name!r} for {model_class!r}"
        )

    field = ContentTypeField(
        to=self.content_type,
        related_name=related_name,
        on_delete=CASCADE,
        no_constraint=(
            getattr(self.content_type, "no_constraint", False)
            or getattr(model_class.meta, "is_tenant", False)
        ),
    )
    # ContentType is managed by registry pre-save hooks.
    field.validator.read_only = True
    field.name = "content_type"
    field.owner = model_class
    field.registry = self
    model_class.fields["content_type"] = field
    model_class.meta.fields["content_type"] = field
    model_class.meta.fields_mapping["content_type"] = field
    model_class.meta.foreign_key_fields["content_type"] = field

    model_related_names = _set_related_name_for_foreign_keys(
        {"content_type": field},
        cast(Any, model_class),
    )
    model_class.meta.related_names.update(model_related_names)

    if (
        getattr(model_class.meta, "registry", None)
        is not getattr(self.content_type.meta, "registry", None)
        or getattr(model_class, "database", None)
        is not getattr(self.content_type, "database", None)
        or getattr(model_class.meta, "is_tenant", False)
    ):
        self.content_type.__require_model_based_deletion__ = True

    self._bind_content_type_pre_save(model_class)

    self._clear_model_table_cache(model_class)
    model_class.__proxy_model__ = None
    proxy_model = model_class.generate_proxy_model()
    model_class.__proxy_model__ = proxy_model
    model_class.__proxy_model__.parent = model_class

def _clear_model_table_cache(self, model_class: type[Any]) -> None:
    """Invalidate cached SQLAlchemy tables for one model.

    Content-type injection and similar runtime mutations change the effective
    table definition. Rather than mutating SQLAlchemy tables in place, the
    registry drops any cached table objects from the primary metadata and
    schema-specific metadata caches so the next `build()` call recreates
    them from the updated model definition.

    Args:
        model_class: Model whose cached tables should be removed.
    """
    model_class._table = None
    model_class._db_schemas = {}

    table_name = cast("str | None", getattr(model_class.meta, "tablename", None))
    if table_name is None:
        return

    metadata_pool = [self._metadata]
    metadata_pool.extend(getattr(self, "_schema_metadata_cache", {}).values())

    for metadata in metadata_pool:
        table_keys = {table_name}
        if metadata.schema:
            table_keys.add(f"{metadata.schema}.{table_name}")
        for table_key in table_keys:
            existing_table = metadata.tables.get(table_key)
            if existing_table is not None:
                metadata.remove(existing_table)

def _bind_content_type_pre_save(self, model_class: type[Any]) -> None:
    """Register a pre-save hook that guarantees content-type rows exist.

    Saffier content types are created lazily. Before a model instance is
    saved, the bound hook inspects every content-type field on the sender and
    creates the matching `ContentType` row when the field is missing or still
    points to an unsaved object.

    Args:
        model_class: Model class that should receive the hook.
    """
    if model_class.__name__ in self._content_type_models_bound:
        return
    if self.content_type is None:
        return
    from saffier.contrib.contenttypes.fields import ContentTypeField

    async def ensure_content_type(
        sender: type[Any],
        instance: Any,
        **kwargs: Any,
    ) -> None:
        if self.content_type is None:
            return
        for field_name, field in sender.fields.items():
            if not isinstance(field, ContentTypeField):
                continue
            current_content_type = instance.__dict__.get(field_name)
            if current_content_type is None and field.null:
                continue
            if (
                current_content_type is not None
                and getattr(current_content_type, "pk", None) is not None
            ):
                continue
            payload = {}
            if current_content_type is not None and hasattr(
                current_content_type, "extract_db_fields"
            ):
                payload = current_content_type.extract_db_fields()

            payload["name"] = sender.__name__
            payload["schema_name"] = instance.get_active_instance_schema()
            content_type_obj = await self.content_type.query.create(**payload)
            setattr(instance, field_name, content_type_obj)

    model_class.signals.pre_save.connect(ensure_content_type)
    self._content_type_models_bound.add(model_class.__name__)

async def create_all(
    self,
    refresh_metadata: bool = True,
    databases: Sequence[str | None] = (None,),
) -> None:
    """Create all tables registered in the target databases.

    Args:
        refresh_metadata: Whether to rebuild registry metadata before
            creating tables.
        databases: Database names to operate on. `None` refers to the
            primary database.
    """
    self._attach_content_type_to_registered_models()
    if refresh_metadata:
        await self.arefresh_metadata(multi_schema=True)
    await self.schema.create_schema(
        self.db_schema,
        if_not_exists=True,
        init_models=True,
        update_cache=bool(self.db_schema),
        databases=databases,
    )

async def drop_all(self, databases: Sequence[str | None] = (None,)) -> None:
    """Drop the registry schema and registered tables from selected databases.

    Args:
        databases: Database aliases to operate on. `None` targets the primary
            database.
    """
    await self.schema.drop_schema(
        self.db_schema,
        cascade=True,
        if_exists=True,
        databases=databases,
    )

def _iter_databases(self) -> list[tuple[str | None, Database]]:
    databases: list[tuple[str | None, Database]] = [(None, self.database)]
    for name, db in self.extra.items():
        databases.append((name, db))
    return databases

def get_model(
    self,
    model_name: str,
    *,
    include_content_type_attr: bool = True,
    include_reflected: bool = True,
    include_pattern: bool = False,
) -> Any:
    """Resolve a model by name from the registry.

    Resolution checks normal registered models first, then reflected models,
    and optionally pattern-generated models. If the stored class is a proxy
    model created for SQLAlchemy compatibility, the parent model is returned
    instead so callers always receive the public ORM model class.

    Args:
        model_name: Name of the model to resolve.
        include_content_type_attr: Whether the synthetic `ContentType`
            attribute should be considered.
        include_reflected: Whether reflected models should be searched.
        include_pattern: Whether pattern-generated reflected models should
            be searched.

    Returns:
        Any: The resolved model class.

    Raises:
        LookupError: If no matching model exists in the configured sources.

    Examples:
        Resolve a model declared elsewhere in the project:

        >>> registry.get_model("User")
    """
    if (
        include_content_type_attr
        and model_name == "ContentType"
        and self.content_type is not None
    ):
        return self.content_type
    if model_name in self.models:
        model = self.models[model_name]
        if getattr(model, "is_proxy_model", False):
            parent = getattr(model, "parent", None)
            if (
                parent is not None
                and getattr(getattr(parent, "meta", None), "registry", None) is self
            ):
                return parent
        return model
    if include_reflected and model_name in self.reflected:
        return self.reflected[model_name]
    if include_pattern and model_name in self.pattern_models:
        return self.pattern_models[model_name]
    raise LookupError(f"Registry doesn't have a {model_name} model.")

def delete_model(self, model_name: str) -> bool:
    """Remove a model from any registry model mapping.

    The helper checks concrete, reflected, and pattern-generated model
    collections. It is used by conflict-resolution code when copying or
    re-registering models.

    Args:
        model_name: Name of the model to remove.

    Returns:
        bool: `True` when a model entry was removed.
    """
    for model_dict in (self.models, self.reflected, self.pattern_models):
        if model_name in model_dict:
            del model_dict[model_name]
            return True
    return False

def register_callback(
    self,
    model_reference: str | type[Any],
    callback: Callable[[type[Any]], None],
    *,
    one_time: bool = False,
) -> None:
    """Register a callback that runs once a model with the given name is available.

    This is used heavily by lazy relation wiring. A field pointing to a
    string model name can register a callback and finish installing reverse
    descriptors as soon as the target model is added to the registry.

    Args:
        model_reference: Target model class or model name to wait for.
        callback: Function called with the resolved model class.
        one_time: When `True`, discard the callback after the first run.
    """
    model_name = (
        model_reference if isinstance(model_reference, str) else model_reference.__name__
    )
    callbacks = self._model_callbacks.setdefault(model_name, [])
    callbacks.append((callback, one_time))

    model_class = self.models.get(model_name) or self.reflected.get(model_name)
    if model_class is None or getattr(model_class, "is_proxy_model", False):
        return
    callback(model_class)
    if one_time:
        callbacks.remove((callback, one_time))
        if not callbacks:
            self._model_callbacks.pop(model_name, None)

def execute_model_callbacks(self, model_class: type[Any]) -> None:
    """Run deferred callbacks waiting for a model to become available.

    Lazy relation wiring registers callbacks keyed by model name. Once the
    target model is registered, this helper executes the callbacks and drops
    one-time entries.

    Args:
        model_class: Newly available model class.
    """
    if getattr(model_class, "is_proxy_model", False):
        return
    callbacks = list(self._model_callbacks.get(model_class.__name__, ()))
    if not callbacks:
        return

    remaining: list[tuple[Callable[[type[Any]], None], bool]] = []
    for callback, one_time in callbacks:
        callback(model_class)
        if not one_time:
            remaining.append((callback, one_time))

    if remaining:
        self._model_callbacks[model_class.__name__] = remaining
    else:
        self._model_callbacks.pop(model_class.__name__, None)

def init_models(
    self, *, init_column_mappers: bool = True, init_class_attrs: bool = True
) -> None:
    """Eagerly initialize metadata caches for registered models.

    This is mainly useful for tooling and tests that want all column-mapping
    caches, table properties, and proxy-model attributes prepared up front.

    Args:
        init_column_mappers: Whether to populate column-to-field mappings.
        init_class_attrs: Whether to warm class-level table and proxy caches.
    """
    for model_class in self.models.values():
        model_class.meta.full_init(
            init_column_mappers=init_column_mappers,
            init_class_attrs=init_class_attrs,
        )
    for model_class in self.reflected.values():
        model_class.meta.full_init(
            init_column_mappers=init_column_mappers,
            init_class_attrs=init_class_attrs,
        )

def invalidate_models(self, *, clear_class_attrs: bool = True) -> None:
    """Invalidate metadata caches for all registered and reflected models.

    Args:
        clear_class_attrs: Whether to also clear cached tables, PK metadata,
            and proxy-model caches on the model classes themselves.
    """
    for model_class in self.models.values():
        model_class.meta.invalidate(clear_class_attrs=clear_class_attrs)
    for model_class in self.reflected.values():
        model_class.meta.invalidate(clear_class_attrs=clear_class_attrs)

def get_tablenames(self) -> set[str]:
    """Collect table names across concrete and reflected models.

    Returns:
        set[str]: All known table names currently tracked by the registry.
    """
    tables = set()
    for model_class in self.models.values():
        tables.add(model_class.meta.tablename)
    for model_class in self.reflected.values():
        tables.add(model_class.meta.tablename)
    return tables

def _automigrate_update(self, migration_settings: Any) -> None:
    from saffier.cli.base import upgrade

    self._is_automigrated = True
    with _monkay.with_full_overwrite(
        extensions={},
        settings=migration_settings,
        instance=Instance(registry=self),
        apply_extensions=True,
        evaluate_settings_with={
            "on_conflict": "replace",
            "ignore_import_errors": False,
            "ignore_preload_import_errors": False,
        },
    ):
        upgrade(app=None)

async def _automigrate(self) -> None:
    migration_settings = self._automigrate_config
    if migration_settings is None or not _monkay.settings.allow_automigrations:
        self._is_automigrated = True
        return
    await asyncio.to_thread(self._automigrate_update, migration_settings)

async def reflect_pattern_models(
    self,
    *,
    database_name: str | None = None,
    database: Database | None = None,
) -> None:
    """Instantiate pattern-based reflected models from live database tables.

    Pattern models act as templates that decide which reflected tables should
    become real runtime models. This method reflects matching schemas, tests
    include and exclude patterns, validates field compatibility, and creates
    concrete reflected models for every matching table exactly once per
    database alias.

    Args:
        database_name: Optional alias of the database being reflected.
        database: Explicit database object to use instead of looking it up
            from the alias.

    Raises:
        RuntimeError: If a reflected pattern would generate a model name that
            conflicts with an existing model.
    """
    if not self.pattern_models:
        return
    if database_name in self._pattern_reflected_dbs:
        return

    target_db = database
    if target_db is None:
        target_db = self.database if database_name is None else self.extra[database_name]

    schemes: set[None | str] = set()
    patterns = []
    for pattern_model in self.pattern_models.values():
        meta = pattern_model.meta
        if database_name not in meta.databases:
            continue
        schemes.update(meta.schemes)
        patterns.append(pattern_model)

    if not patterns:
        self._pattern_reflected_dbs.add(database_name)
        return

    tmp_metadata = sqlalchemy.MetaData()
    for schema in schemes:
        await target_db.run_sync(self._reflect_schema_metadata, tmp_metadata, schema)

    for table in tmp_metadata.tables.values():
        for pattern_model in patterns:
            meta = pattern_model.meta
            if table.schema not in meta.schemes:
                continue
            if not meta.include_pattern.match(table.name):
                continue
            if meta.exclude_pattern and meta.exclude_pattern.match(table.name):
                continue
            if pattern_model.fields_not_supported_by_table(table):
                continue

            model_name = meta.template(table)
            try:
                self.get_model(model_name, include_pattern=False)
            except LookupError:
                ...
            else:
                raise RuntimeError(
                    f"Conflicting reflected model name generated: {model_name!r}."
                )

            pattern_model.create_reflected_model(
                table=table,
                registry=self,
                database=target_db,
                name=model_name,
            )

    self._pattern_reflected_dbs.add(database_name)