Queries¶

class Query(Generic[T]):
    """Build and execute fluent ORM queries.

    Attributes:
        model_cls: Model class used to hydrate results.
        where_clause: Accumulated filter nodes for the query.
        order_by_clause: Sort definitions sent to the Rust core.
    """

    def __init__(
        self, model_cls: Type[T], using: str | None = None, session: Any | None = None
    ):
        """Initialize a query for a model class.

        Args:
            model_cls: Model class that defines the target table.

        Examples:
            >>> query = Query(User)
            >>> query.model_cls is User
            True
        """
        self.model_cls = model_cls
        self._using = using
        self._session = session
        self.where_clause: list["QueryNode"] = []
        self.order_by_clause: list[dict[str, str]] = []
        self._limit: int | None = None
        self._offset: int | None = None
        self._m2m_context: dict[str, Any] | None = None

    def _transaction_or_using(self) -> tuple[str | None, str | None, str | None]:
        from ..state import resolve_operation_scope

        return resolve_operation_scope(
            using=self._using, session=self._session, allow_legacy_default=True
        )

    def _m2m(
        self, join_table: str, source_col: str, target_col: str, source_id: Any
    ) -> "Query[T]":
        """Store many-to-many linkage context for relationship operations"""
        self._m2m_context = {
            "join_table": join_table,
            "source_col": source_col,
            "target_col": target_col,
            "source_id": source_id,
        }
        return self

    @overload
    def where(self, node: "QueryNode") -> "Query[T]": ...

    @overload
    def where(self, node: "Predicate[T]") -> "Query[T]": ...

    def where(self, node: "QueryNode | Predicate[T]") -> "Query[T]":
        """Add a filter condition to the query.

        The recommended style is a lambda predicate of shape
        ``Callable[[QueryProxy[T]], QueryNode]``. The lambda receives a
        fresh :class:`QueryProxy` whose attributes return
        :class:`FieldProxy` instances, so ``lambda t: t.archived == False``
        builds a comparison without static-typing friction. A prebuilt
        :class:`QueryNode` is also accepted, built either with
        :func:`ferro.query.col` (the type-safe escape hatch that preserves
        operator shape) or with operator syntax on class attributes. The
        bare operator form (``User.where(User.age >= 18)``) is deprecated and
        on the v0.14.0 removal track. It does not
        type-check statically:
        the class attribute types as the field type, so the comparison
        resolves to ``bool``, not ``QueryNode``.

        Args:
            node: A predicate callable or a ``QueryNode``.

        Returns:
            The current Query instance for chaining.

        Raises:
            TypeError: If ``node`` is neither a ``QueryNode`` nor a callable,
                or if the callable does not return a ``QueryNode``.

        Examples:
            >>> q1 = User.where(lambda t: t.archived == False)  # noqa: E712
            >>> q2 = User.where(lambda t: t.id == 1)
            >>> isinstance(q1, Query) and isinstance(q2, Query)
            True
        """
        self.where_clause.append(_resolve_where_node(node))
        return self

    def order_by(self, field: Any, direction: str = "asc") -> "Query[T]":
        """Add an ordering clause to the query

        Args:
            field: The field to order by (e.g., User.username).
            direction: The direction of the sort ("asc" or "desc").

        Returns:
            The current Query instance for chaining.

        Raises:
            ValueError: If direction is not "asc" or "desc".

        Examples:
            >>> query = User.select().order_by(User.username, "desc")
            >>> query.order_by_clause[-1]["direction"]
            'desc'
        """
        if direction.lower() not in ("asc", "desc"):
            raise ValueError("direction must be 'asc' or 'desc'")

        col_name = field.column if hasattr(field, "column") else str(field)
        self.order_by_clause.append(
            {"column": col_name, "direction": direction.lower()}
        )
        return self

    def limit(self, value: int) -> "Query[T]":
        """Limit the number of records returned

        Args:
            value: The maximum number of records to return.

        Returns:
            The current Query instance for chaining.

        Examples:
            >>> query = User.select().limit(10)
            >>> query._limit
            10
        """
        self._limit = value
        return self

    def offset(self, value: int) -> "Query[T]":
        """Skip a specific number of records

        Args:
            value: The number of records to skip.

        Returns:
            The current Query instance for chaining.

        Examples:
            >>> query = User.select().offset(20)
            >>> query._offset
            20
        """
        self._offset = value
        return self

    async def all(self) -> list[T]:
        """Return all model instances that match the current query

        Returns:
            A list of model instances.

        Examples:
            >>> users = await User.where(lambda t: t.active == True).all()  # noqa: E712
            >>> isinstance(users, list)
            True
        """
        query_def = {
            "model_name": self.model_cls.__name__,
            "where": [node.to_ir_dict() for node in self.where_clause],
            "order_by": self.order_by_clause,
            "limit": self._limit,
            "offset": self._offset,
            "m2m": self._m2m_context,
        }
        tx_id, using, session_id = self._transaction_or_using()
        results = await fetch_filtered(
            self.model_cls,
            _query_ir_payload_to_json(query_def),
            tx_id,
            using,
            session_id=session_id,
        )
        for instance in results:
            if hasattr(self.model_cls, "_fix_types"):
                self.model_cls._fix_types(instance)
        return results

    async def count(self) -> int:
        """Return the number of records that match the current query

        Returns:
            The count of matching records.

        Examples:
            >>> total = await User.where(lambda t: t.active == True).count()  # noqa: E712
            >>> isinstance(total, int)
            True
        """
        query_def = {
            "model_name": self.model_cls.__name__,
            "where": [node.to_ir_dict() for node in self.where_clause],
            "order_by": [],
            "limit": None,
            "offset": None,
            "m2m": self._m2m_context,
        }
        tx_id, using, session_id = self._transaction_or_using()
        return await count_filtered(
            self.model_cls.__name__,
            _query_ir_payload_to_json(query_def),
            tx_id,
            using,
            session_id=session_id,
        )

    async def update(self, **fields) -> int:
        """Update all records matching the current query

        Args:
            **fields: Field names and values to update.

        Returns:
            The number of records updated.

        Examples:
            >>> updated = await User.where(lambda t: t.id == 1).update(name="Taylor")
            >>> isinstance(updated, int)
            True
        """
        query_def = {
            "model_name": self.model_cls.__name__,
            "where": [node.to_ir_dict() for node in self.where_clause],
            "order_by": [],
            "limit": self._limit,
            "offset": self._offset,
            "m2m": None,
        }
        from pydantic_core import to_json

        tx_id, using, session_id = self._transaction_or_using()
        # Use pydantic_core.to_json to handle Decimals, UUIDs, etc. in kwargs
        return await update_filtered(
            self.model_cls.__name__,
            _query_ir_payload_to_json(query_def),
            to_json(fields).decode(),
            tx_id,
            using,
            session_id=session_id,
        )

    async def first(self) -> T | None:
        """Return the first matching record, or None

        Returns:
            A model instance or None.

        Examples:
            >>> user = await User.select().order_by(User.id).first()
            >>> user is None or isinstance(user, User)
            True
        """
        old_limit = self._limit
        self._limit = 1
        try:
            results = await self.all()
            return results[0] if results else None
        finally:
            self._limit = old_limit

    async def delete(self) -> int:
        """Delete all records matching the current query

        Returns:
            The number of records deleted.

        Examples:
            >>> deleted = await User.where(lambda t: t.disabled == True).delete()  # noqa: E712
            >>> isinstance(deleted, int)
            True
        """
        query_def = {
            "model_name": self.model_cls.__name__,
            "where": [node.to_ir_dict() for node in self.where_clause],
            "order_by": [],
            "limit": self._limit,
            "offset": self._offset,
            "m2m": None,
        }
        tx_id, using, session_id = self._transaction_or_using()
        return await delete_filtered(
            self.model_cls.__name__,
            _query_ir_payload_to_json(query_def),
            tx_id,
            using,
            session_id=session_id,
        )

    async def exists(self) -> bool:
        """Return whether at least one record matches the current query

        Returns:
            True if records exist, otherwise False.

        Examples:
            >>> found = await User.where(lambda t: t.email == "a@b.com").exists()
            >>> isinstance(found, bool)
            True
        """
        return await self.count() > 0

    async def add(self, *instances: Any) -> None:
        """Add links to a many-to-many relationship

        Args:
            *instances: Target model instances that provide an ``id`` attribute.

        Raises:
            RuntimeError: If the query is not bound to a many-to-many context.

        Examples:
            >>> user = await User.create(email="taylor@example.com")
            >>> admin = await Group.create(name="admin")
            >>> staff = await Group.create(name="staff")
            >>> await user.groups.add(admin, staff)
        """
        if not self._m2m_context:
            raise RuntimeError(
                "'.add()' can only be used on Many-to-Many relationships"
            )

        ids = []
        for inst in instances:
            # Assume 'id' for now
            ids.append(getattr(inst, "id"))

        from ..state import _CURRENT_TRANSACTION

        tx_id, using, session_id = self._transaction_or_using()
        await add_m2m_links(
            self._m2m_context["join_table"],
            self._m2m_context["source_col"],
            self._m2m_context["target_col"],
            self._m2m_context["source_id"],
            ids,
            tx_id,
            using,
            session_id=session_id,
        )

    async def remove(self, *instances: Any) -> None:
        """Remove links from a many-to-many relationship

        Args:
            *instances: Target model instances that provide an ``id`` attribute.

        Raises:
            RuntimeError: If the query is not bound to a many-to-many context.

        Examples:
            >>> user = await User.create(email="taylor@example.com")
            >>> admin = await Group.create(name="admin")
            >>> await user.groups.remove(admin)
        """
        if not self._m2m_context:
            raise RuntimeError(
                "'.remove()' can only be used on Many-to-Many relationships"
            )

        ids = []
        for inst in instances:
            ids.append(getattr(inst, "id"))

        from ..state import _CURRENT_TRANSACTION

        tx_id, using, session_id = self._transaction_or_using()
        await remove_m2m_links(
            self._m2m_context["join_table"],
            self._m2m_context["source_col"],
            self._m2m_context["target_col"],
            self._m2m_context["source_id"],
            ids,
            tx_id,
            using,
            session_id=session_id,
        )

    async def clear(self) -> None:
        """Clear all links in a many-to-many relationship

        Raises:
            RuntimeError: If the query is not bound to a many-to-many context.

        Examples:
            >>> user = await User.create(email="taylor@example.com")
            >>> await user.groups.clear()
        """
        if not self._m2m_context:
            raise RuntimeError(
                "'.clear()' can only be used on Many-to-Many relationships"
            )

        from ..state import _CURRENT_TRANSACTION

        tx_id, using, session_id = self._transaction_or_using()
        await clear_m2m_links(
            self._m2m_context["join_table"],
            self._m2m_context["source_col"],
            self._m2m_context["source_id"],
            tx_id,
            using,
            session_id=session_id,
        )

    def __repr__(self):
        """Return a developer-friendly representation of the query"""
        return f"<Query model={self.model_cls.__name__} where={self.where_clause}>"

class QueryProxy(Generic[TModel]):
    """Lazy attribute proxy used by lambda predicates passed to ``Query.where``.

    A fresh ``QueryProxy`` is constructed each time a lambda predicate is
    evaluated. Any attribute access returns a :class:`FieldProxy` for the
    accessed name, so ``lambda t: t.archived == False`` builds a
    :class:`QueryNode` without ever asking the model class what type
    ``archived`` is. The ``TModel`` type parameter exists so user-supplied
    lambdas can narrow ``t`` to a specific model in static analysis; the
    proxy itself ignores the parameter at runtime.

    The proxy attribute return type is intentionally ``FieldProxy[Any]`` for
    now — wiring per-field types through a lambda parameter requires
    ``@dataclass_transform`` plumbing on the metaclass, which is outside this
    feature's scope.

    Examples:
        >>> rows = await User.where(lambda t: t.archived == False).all()  # noqa: E712
    """

    __slots__ = ()

    def __getattr__(self, name: str) -> "FieldProxy[Any]":
        """Return a fresh ``FieldProxy`` for any attribute name."""
        return FieldProxy(name, predicate_style="lambda")