Registry¶

async def apply_default_force_nullable_fields(
    self,
    *,
    force_fields_nullable: Iterable[tuple[str, str]] | None = None,
    model_defaults: dict[str, dict[str, Any]] | None = None,
    filter_db_url: str | None = None,
    filter_db_name: str | None = None,
) -> None:
    """Backfill defaults for fields made temporarily nullable by migrations.

    Forced-nullable migration generation is a two-step safety mechanism.
    Alembic first sees selected fields as nullable so it can add them to
    populated tables without violating constraints. During online upgrade,
    generated migration files call this method to find rows where those
    fields are still ``NULL`` and update them with the model's declared
    defaults or with explicit defaults supplied by the migration.

    The method deliberately stays on Saffier's public ORM path. It filters
    registered models, selects only primary-key data for matching rows, then
    calls ``update()`` on each model instance with concrete default values.
    That preserves field normalization, SQLAlchemy execution, and model
    update signals while avoiding a second migration-specific persistence
    layer.

    Args:
        force_fields_nullable: Iterable of ``(model_name, field_name)``
            selectors. When omitted, selectors are read from the active
            ``FORCE_FIELDS_NULLABLE`` context.
        model_defaults: Optional per-model default overrides keyed by model
            name and field name. Overrides are also treated as selectors so
            migrations can backfill fields that do not declare a Python
            default on the model.
        filter_db_url: Optional database URL string. When provided, only
            models using that database are processed.
        filter_db_name: Optional registry database name. ``""`` targets the
            primary database; any other value targets the matching
            ``extra`` database and takes precedence over ``filter_db_url``.
    """
    if force_fields_nullable is None:
        selected_fields = set(FORCE_FIELDS_NULLABLE.get())
    else:
        selected_fields = set(force_fields_nullable)
    effective_model_defaults = model_defaults or {}
    for model_name, defaults in effective_model_defaults.items():
        for field_name in defaults:
            selected_fields.add((model_name, field_name))

    for model_name, field_name in tuple(selected_fields):
        if model_name:
            continue
        selected_fields.discard((model_name, field_name))
        for model_class in self.models.values():
            if field_name in model_class.fields:
                selected_fields.add((model_class.__name__, field_name))

    if not selected_fields:
        return

    if isinstance(filter_db_name, str):
        filter_db_url = (
            str(self.database.url)
            if filter_db_name == ""
            else str(self.extra[filter_db_name].url)
        )

    models_with_fields: dict[str, set[str]] = {}
    for model_name, field_name in selected_fields:
        model_class = self.models.get(model_name)
        if model_class is None:
            continue

        field = model_class.fields.get(field_name)
        if field is None:
            continue

        default_overrides = effective_model_defaults.get(model_name, {})
        if not field.has_default() and field_name not in default_overrides:
            continue

        models_with_fields.setdefault(model_name, set()).add(field_name)

    async def backfill_model(
        model_name: str,
        field_names: set[str],
    ) -> None:
        """Backfill one model's matching rows through normal ORM updates.

        The closure is scheduled by ``run_concurrently`` for each model that
        has at least one valid forced-nullable field. It keeps row selection
        narrow by loading primary keys first, then computes default values
        per object so callable defaults are evaluated for each updated row
        rather than once for the whole migration.

        Args:
            model_name: Registered model class name to process.
            field_names: Field names that were temporarily nullable and
                should now receive defaults where the database still stores
                ``NULL``.
        """
        model_class = self.models[model_name]
        if filter_db_url and str(model_class.database.url) != filter_db_url:
            return

        query = model_class.query.filter(**dict.fromkeys(field_names)).only(
            *model_class.pknames
        )
        objects = await query
        default_overrides = effective_model_defaults.get(model_name, {})
        for obj in objects:
            values = {
                field_name: (
                    default_overrides[field_name]
                    if field_name in default_overrides
                    else model_class.fields[field_name].get_default_value()
                )
                for field_name in field_names
            }
            await obj.update(**values)

    await run_concurrently(
        [
            backfill_model(model_name, field_names)
            for model_name, field_names in models_with_fields.items()
        ]
    )

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)
    registry_copy.admin_models = {
        name
        for name in self.admin_models
        if name in registry_copy.models or name in registry_copy.reflected
    }
    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):
        in_admin = getattr(content_type_model.meta, "in_admin", None)
        no_admin_create = getattr(content_type_model.meta, "no_admin_create", None)
        meta = type(
            "Meta",
            (),
            {
                "registry": self,
                "tablename": "contenttypes",
                "in_admin": True if in_admin is None else in_admin,
                "no_admin_create": True if no_admin_create is None else no_admin_create,
            },
        )
        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"
        if getattr(content_type_model.meta, "in_admin", None) is None:
            content_type_model.meta.in_admin = True
        if getattr(content_type_model.meta, "no_admin_create", None) is None:
            content_type_model.meta.no_admin_create = True
        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:
    """Finalize registry-owned state for a newly registered model.

    Model registration is responsible for more than adding the class to
    ``registry.models``. The registry also owns the admin visibility index
    and optional content-type wiring. Updating both concerns here keeps
    dynamically registered, copied, and generated models consistent after
    their concrete registry name is known.

    Args:
        model_class: Model class that has just been registered.
    """
    if not getattr(model_class, "is_proxy_model", False) and not getattr(
        model_class.meta,
        "abstract",
        False,
    ):
        model_name = model_class.__name__
        if getattr(model_class.meta, "in_admin", None) is not False:
            self.admin_models.add(model_name)
        else:
            self.admin_models.discard(model_name)

    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. The admin model index is updated at the same time
    so deleted models cannot remain visible in the built-in admin.

    Args:
        model_name: Name of the model to remove.

    Returns:
        bool: `True` when a model entry was removed.
    """
    self.admin_models.discard(model_name)
    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,
        )