Model¶

class Model(BaseModel, metaclass=ModelMetaclass):
    """Provide the base class for all Ferro models

    Inheriting from this class registers schema metadata with the Rust core and
    exposes high-performance CRUD and query entrypoints.

    **Composite unique constraints:** declare a ``typing.ClassVar`` named
    ``__ferro_composite_uniques__`` as a tuple of tuples of column names
    (for example ``(("user_id", "org_id"),)``) to enforce uniqueness on those
    columns together. This is separate from per-column uniqueness
    (``Field(unique=True)`` on the field, ``Annotated[..., Field(unique=True)]``,
    or ``Annotated[..., FerroField(unique=True)]``), each of which applies to a
    single column only. Default many-to-many join tables get a
    composite unique on their two foreign-key columns automatically.

    **Composite indexes:** declare a ``typing.ClassVar`` named
    ``__ferro_composite_indexes__`` as a tuple of tuples of column names
    (for example ``(("user_id", "created_at"),)``) for non-unique multi-column
    indexes. Validation rules mirror ``__ferro_composite_uniques__``: each
    inner tuple must contain at least two columns, columns must exist on the
    model, and order is preserved (matters for leftmost-prefix optimization).
    For single-column indexes use ``Field(index=True)``. Default many-to-many
    join tables get a non-unique reverse-direction composite index
    automatically; opt out with ``ManyToMany(reverse_index=False)``.

    Examples:
        >>> class User(Model):
        ...     id: int | None = None
        ...     name: str
    """

    __ferro_composite_uniques__: ClassVar[tuple[tuple[str, ...], ...]] = ()
    __ferro_composite_indexes__: ClassVar[tuple[tuple[str, ...], ...]] = ()
    _enum_fields: ClassVar[dict[str, type[Enum]]] = {}

    @classmethod
    def _reregister_ferro(cls) -> None:
        """Re-register this model's schema with the Rust core (e.g. after clear_registry)."""
        schema = getattr(cls, "__ferro_schema__", None)
        if schema is not None:
            from ._core import register_model_schema

            register_model_schema(
                cls.__ferro_identity__, json.dumps(schema), cls.__ferro_table__
            )

    model_config = ConfigDict(
        from_attributes=True,
        use_attribute_docstrings=True,
        arbitrary_types_allowed=True,
    )

    def __init__(self, **data: Any):
        """Initialize a model instance and normalize relationship inputs

        Args:
            **data: Field values used to construct the model.

        Examples:
            >>> user = User(name="Taylor")
            >>> isinstance(user, User)
            True
        """
        # 1. Handle relationship inputs (e.g. Product(category=my_cat))
        relations = getattr(self.__class__, "ferro_relations", {})
        for field_name, metadata in relations.items():
            if isinstance(metadata, ForeignKey) and field_name in data:
                val = data.pop(field_name)
                # If it's a Model instance, extract the ID
                if isinstance(val, Model):
                    # Read the *target* model's PK (FF-D D5) — the source
                    # model's PK name is irrelevant to the related instance.
                    pk_field = val.__class__._primary_key_field_name() or "id"
                    id_val = getattr(val, pk_field, None)
                    data[f"{field_name}_id"] = id_val
                else:
                    # It's already an ID or something else
                    data[f"{field_name}_id"] = val

        super().__init__(**data)

    @model_validator(mode="after")
    def _validate_required_foreign_keys(self) -> Self:
        """Keep Python model validation aligned with required FK nullability."""
        relations = getattr(self.__class__, "ferro_relations", {})
        for field_name, metadata in relations.items():
            if not isinstance(metadata, ForeignKey):
                continue
            if foreign_key_allows_none(metadata) is False:
                if getattr(self, f"{field_name}_id", None) is None:
                    raise ValueError(f"{field_name} is required")
        return self

    async def save(
        self,
        *,
        using: str | None = None,
        session: "Session | None" = None,
        on_conflict: Literal["update"] | None = None,
    ) -> None:
        """Persist the current model instance.

        A transient instance (constructed with ``Model(...)`` and never saved)
        is INSERTed — a duplicate primary key or unique value raises
        :class:`~ferro.exceptions.UniqueViolationError`. A persistent instance
        (fetched from the database, or previously saved) is UPDATEd by primary
        key. Pass ``on_conflict="update"`` for insert-or-update semantics
        regardless of persistence state (the primitive behind
        :meth:`upsert`).

        Note that ``model_copy()`` copies persistence state: saving a copy of
        a persisted instance updates the same row. The UPDATE targets the
        instance's *current* primary-key value, so mutating the PK of a
        persisted instance before ``save()`` matches no row and raises. A row
        inserted inside a rolled-back transaction leaves the instance marked
        persisted; a later ``save()`` raises ``ModelDoesNotExist``.

        Args:
            using: Connection name override.
            session: Session scope for the operation.
            on_conflict: ``None`` (default) or ``"update"`` to upsert.

        Raises:
            UniqueViolationError: A duplicate primary key or unique value on
                INSERT.
            ModelDoesNotExist: The row behind a persisted instance no longer
                exists (deleted underneath, or the PK was mutated).
            ValueError: ``on_conflict`` is not ``None`` or ``"update"``, or a
                persisted instance has no primary-key value.

        Examples:
            >>> user = User(name="Taylor")
            >>> await user.save()
        """
        if on_conflict not in (None, "update"):
            raise ValueError(
                f'on_conflict must be None or "update", got {on_conflict!r}'
            )
        route, identity_using = _instance_transaction_route(self, using, session)
        new_id = None
        if on_conflict == "update":
            new_id = await save_record(
                self.__class__.__ferro_identity__,
                save_bind_payload(self),
                route,
                mode="upsert",
            )
        elif _is_persisted(self):
            pk_field_name = self.__class__._primary_key_field_name()
            pk_val = getattr(self, pk_field_name) if pk_field_name is not None else None
            if pk_val is None:
                raise ValueError(
                    f"Cannot UPDATE a persisted {self.__class__.__name__} "
                    "without a primary key value"
                )
            rows_affected = await update_record(
                self.__class__.__ferro_identity__,
                save_bind_payload(self),
                route,
            )
            if rows_affected == 0:
                raise ModelDoesNotExist(self.__class__, pk_val)
        else:
            new_id = await save_record(
                self.__class__.__ferro_identity__,
                save_bind_payload(self),
                route,
                mode="insert",
            )

        pk_val = None
        pk_field_name = None

        for field_name, metadata in self.__class__.ferro_fields.items():
            if metadata.primary_key:
                pk_field_name = field_name
                if metadata.autoincrement and getattr(self, field_name) is None:
                    if new_id is not None:
                        setattr(self, field_name, new_id)
                pk_val = getattr(self, field_name)
                break

        if pk_field_name is None:
            for field_name, field in self.__class__.model_fields.items():
                if getattr(field, "json_schema_extra", {}).get("primary_key"):
                    pk_field_name = field_name
                    if getattr(self, field_name) is None and new_id is not None:
                        setattr(self, field_name, new_id)
                    pk_val = getattr(self, field_name)
                    break

        if pk_val is not None:
            register_instance(
                self.__class__.__ferro_identity__,
                str(pk_val),
                self,
                route,
            )
            _set_instance_origin(self, identity_using)
        _set_persisted(self, True)

    async def delete(
        self, *, using: str | None = None, session: "Session | None" = None
    ) -> None:
        """Delete the current model instance from storage

        Returns:
            None

        Examples:
            >>> user = await User.get_or_none(1)
            >>> if user:
            ...     await user.delete()
        """
        pk_field_name = self.__class__._primary_key_field_name()
        pk_val = getattr(self, pk_field_name) if pk_field_name is not None else None
        route, _identity_using = _instance_transaction_route(self, using, session)

        if pk_val is not None:
            name = self.__class__.__ferro_identity__
            query = Query(self.__class__, using=route.connection_name).where(
                _field_eq(pk_field_name, pk_val)
            )
            await query.delete()
            _core_evict_instance(name, str(pk_val), route)
            # The instance is transient again: a later save() re-INSERTs.
            _set_persisted(self, False)

    @classmethod
    def _primary_key_field_name(cls) -> str | None:
        for field_name, metadata in cls.ferro_fields.items():
            if metadata.primary_key:
                return field_name

        for field_name, field in cls.model_fields.items():
            if getattr(field, "json_schema_extra", {}).get("primary_key"):
                return field_name

        return None

    @classmethod
    async def all(
        cls, *, using: str | None = None, session: "Session | None" = None
    ) -> list[Self]:
        """Fetch all records for this model class

        Returns:
            A list of hydrated model instances.

        Examples:
            >>> users = await User.all()
            >>> isinstance(users, list)
            True
        """
        route = _transaction_or_using(using, session)
        return await fetch_all(cls, route)

    @classmethod
    async def get(cls, pk: Any, *, session: "Session | None" = None) -> Self:
        """Fetch one record by primary key value.

        Args:
            pk: Primary key value to fetch a single record.

        Returns:
            The matching model instance.

        Raises:
            ModelDoesNotExist: When no row exists for this primary key. Use
                :meth:`get_or_none` if you need optional lookup without raising.

        Examples:
            >>> user = await User.get(1)
            >>> isinstance(user, User)
            True
        """
        instance = await cls.get_or_none(pk, session=session)
        if instance is None:
            raise ModelDoesNotExist(cls, pk)
        return instance

    @classmethod
    async def get_or_none(
        cls, pk: Any, *, session: "Session | None" = None
    ) -> Self | None:
        """Fetch one record by primary key, or return None if no row exists.

        Args:
            pk: Primary key value to fetch a single record.

        Returns:
            The matching model instance, or None when no record exists.
        """
        pk_field_name = cls._primary_key_field_name()
        if pk_field_name is None:
            raise RuntimeError(f"Model {cls.__name__} does not define a primary key")

        return await cls.where(_field_eq(pk_field_name, pk), session=session).first()

    async def refresh(
        self, *, using: str | None = None, session: "Session | None" = None
    ) -> None:
        """Reload this instance from storage using its primary key

        Returns:
            None

        Raises:
            RuntimeError: If no primary key is available or the record no longer exists.

        Examples:
            >>> user = await User.get(1)
            >>> await user.refresh()
        """
        pk_field_name = self.__class__._primary_key_field_name()
        pk_val = getattr(self, pk_field_name) if pk_field_name is not None else None

        if pk_val is None:
            raise RuntimeError("Cannot refresh a model without a primary key")

        name = self.__class__.__ferro_identity__
        route, identity_using = _instance_transaction_route(self, using, session)

        _core_evict_instance(name, str(pk_val), route)
        query = Query(self.__class__, using=route.connection_name).where(
            _field_eq(pk_field_name, pk_val)
        )
        fresh_instance = await query.first()

        if fresh_instance is None:
            raise RuntimeError(f"Instance not found in database: {name}({pk_val})")

        self.__dict__.update(fresh_instance.__dict__)
        register_instance(name, str(pk_val), self, route)
        _set_instance_origin(self, identity_using)
        _set_persisted(self, True)

    @classmethod
    def where(
        cls, predicate: "Predicate[Self]", *, session: "Session | None" = None
    ) -> Query[Self]:
        """Start a fluent query with an initial condition.

        ``predicate`` is a lambda of shape
        ``Callable[[QueryProxy[Self]], QueryNode]``, e.g.
        ``User.where(lambda user: user.age >= 18)``. The lambda receives a
        :class:`QueryProxy` whose attributes build comparisons as
        :class:`QueryNode` instances, so predicates type-check cleanly.
        Name the parameter after the model in lowercase singular (``user`` for
        ``User``, ``post`` for ``Post``). Column names are validated at build
        time against the model's declared fields (plus shadow ``{fk}_id``
        columns).

        Args:
            predicate: A callable that takes a :class:`QueryProxy` and
                returns a :class:`QueryNode`.

        Returns:
            A query object scoped to this model class.

        Examples:
            >>> q1 = User.where(lambda user: user.archived == False)  # noqa: E712
            >>> q2 = User.where(lambda user: user.id == 1)
            >>> isinstance(q1, Query) and isinstance(q2, Query)
            True
        """
        return Query(cls, session=session).where(predicate)

    @classmethod
    def select(cls, *, session: "Session | None" = None) -> Query[Self]:
        """Start an empty fluent query for this model class

        Returns:
            A query object scoped to this model class.

        Examples:
            >>> query = User.select().limit(5)
            >>> isinstance(query, Query)
            True
        """
        return Query(cls, session=session)

    @classmethod
    def using(cls, name: str) -> "ModelConnection[Self]":
        """Bind ORM operations for this model to a named connection."""
        return ModelConnection(cls, name)

    @classmethod
    async def create(cls, *, session: "Session | None" = None, **fields) -> Self:
        """Create and persist a new model instance

        ``create()`` is a plain INSERT: it never updates an existing row.

        Args:
            **fields: Field values to construct the model.

        Returns:
            The newly created and persisted model instance.

        Raises:
            UniqueViolationError: A row with the same primary key or unique
                value already exists — use :meth:`upsert` for
                insert-or-update semantics.

        Examples:
            >>> user = await User.create(name="Taylor")
            >>> isinstance(user, User)
            True
        """
        instance = cls(**fields)
        await instance.save(session=session)
        return instance

    @classmethod
    async def upsert(cls, *, session: "Session | None" = None, **fields) -> Self:
        """Insert the row, or update the existing row on primary-key conflict.

        Equivalent to ``cls(**fields).save(on_conflict="update")``. With an
        autoincrement primary key left unset there is no conflict target, so
        this degrades to a plain INSERT.

        Args:
            **fields: Field values to construct the model.

        Returns:
            The persisted model instance.

        Examples:
            >>> user = await User.upsert(id=1, name="Taylor")
            >>> isinstance(user, User)
            True
        """
        instance = cls(**fields)
        await instance.save(session=session, on_conflict="update")
        return instance

    @classmethod
    async def bulk_create(
        cls,
        instances: list[Self],
        *,
        using: str | None = None,
        session: "Session | None" = None,
    ) -> int:
        """Persist multiple instances in a single bulk operation

        Args:
            instances: Model instances to persist.

        Returns:
            The number of records inserted.

        Examples:
            >>> rows = await User.bulk_create([User(name="A"), User(name="B")])
            >>> isinstance(rows, int)
            True
        """
        if not instances:
            return 0
        data = [save_bind_payload(i) for i in instances]
        route = _transaction_or_using(using, session)
        return await save_bulk_records(cls.__ferro_identity__, data, route)

    @classmethod
    async def get_or_create(
        cls,
        defaults: dict[str, Any] | None = None,
        *,
        session: "Session | None" = None,
        **fields,
    ) -> tuple[Self, bool]:
        """Fetch a record by filters or create one when missing

        Args:
            defaults: Values applied only when creating a new record.
            **fields: Exact-match filters used for lookup.

        Returns:
            A tuple of ``(instance, created)`` where ``created`` is True for new records.

        Examples:
            >>> user, created = await User.get_or_create(email="[email protected]")
            >>> isinstance(created, bool)
            True
        """
        query = Query(cls, session=session)
        for key, val in fields.items():
            query = query.where(_field_eq(key, val))

        instance = await query.first()
        if instance:
            return instance, False

        params = {**fields, **(defaults or {})}
        return await cls.create(session=session, **params), True

    @classmethod
    async def update_or_create(
        cls,
        defaults: dict[str, Any] | None = None,
        *,
        session: "Session | None" = None,
        **fields,
    ) -> tuple[Self, bool]:
        """Update a matched record or create one when missing

        Args:
            defaults: Values applied on update or create paths.
            **fields: Exact-match filters used for lookup.

        Returns:
            A tuple of ``(instance, created)`` where ``created`` is True for new records.
        """
        query = Query(cls, session=session)
        for key, val in fields.items():
            query = query.where(_field_eq(key, val))

        instance = await query.first()
        if instance:
            for key, val in (defaults or {}).items():
                setattr(instance, key, val)
            await instance.save(session=session)
            return instance, False

        params = {**fields, **(defaults or {})}
        return await cls.create(session=session, **params), True