Declarative Repositories & Queries¶

@component
class RepositoryQueryInterceptor(MethodInterceptor):
    """Executes declarative queries for ``@query``-decorated methods.

    This interceptor runs **after** ``TransactionalInterceptor`` in the
    AOP chain.  It checks for ``QUERY_META`` on the target method; if
    absent, it simply delegates to ``call_next``.  When present, the
    method body is **never executed** -- instead the interceptor builds
    and runs the SQL query, binding method parameters automatically.

    Two execution modes:

    * ``expr`` -- generates ``SELECT * FROM <table> WHERE <expr>`` using
      the entity from ``@repository(entity=...)``.
    * ``sql`` -- executes the provided raw SQL verbatim.

    Both modes support pagination (``paged=True``) and unique result
    (``unique=True``).

    Args:
        session_manager: The ``SessionManager`` singleton, used to
            obtain the current ``AsyncSession`` via ``get_session()``.
    """

    def __init__(self, session_manager: SessionManager):
        self.session_manager = session_manager

    async def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        """Intercept the method and execute the declarative query if applicable.

        Args:
            ctx: The AOP method context.
            call_next: Callback to the next interceptor or the original
                method.

        Returns:
            Query results -- a ``Page``, a list of ``RowMapping``, or a
            single ``RowMapping`` / ``None`` depending on query options.

        Raises:
            RuntimeError: ``"@query with expr requires @repository(entity=...)"``
                if expression mode is used without an entity.
            TypeError: ``"Paged query requires a 'page: PageRequest' parameter"``
                if ``paged=True`` but no ``PageRequest`` argument is found.
            ValueError: ``"Dynamic sorting via PageRequest is not supported
                in SQL mode"`` if sorts are provided in SQL mode.
            ValueError: ``"Invalid sort field: <field>"`` if a sort field
                is not a valid column on the entity.
        """
        func = getattr(ctx.cls, ctx.name, None)
        meta = getattr(func, QUERY_META, None)

        if meta is None:
            return await self._call_next_async(ctx, call_next)

        session = get_session(self.session_manager)
        params = self._bind_params(func, ctx.args, ctx.kwargs)
        repo_meta = getattr(ctx.cls, REPOSITORY_META, {}) or {}
        entity = repo_meta.get("entity")
        mode = meta.get("mode")

        if mode == "sql":
            return await self._execute_sql(session, meta, params)
        if mode == "expr":
            return await self._execute_expr(session, meta, params, entity)
        raise RuntimeError(f"Unsupported query mode: {mode!r}")

    async def _call_next_async(self, ctx: MethodCtx, call_next: Callable) -> Any:
        """Invoke the next interceptor or method, awaiting if needed."""
        result = call_next(ctx)
        if inspect.isawaitable(result):
            result = await result
        return result

    def _bind_params(
        self,
        func: Callable[..., Any],
        args: tuple[Any, ...],
        kwargs: Mapping[str, Any],
    ) -> dict[str, Any]:
        """Bind positional and keyword arguments to a named parameter dict.

        Uses ``inspect.signature`` to map *args* and *kwargs* onto the
        method's declared parameters (excluding ``self``).

        Args:
            func: The original method (unbound).
            args: Positional arguments from the invocation.
            kwargs: Keyword arguments from the invocation.

        Returns:
            A ``dict`` mapping parameter names to their values, suitable
            for use as SQLAlchemy bind parameters.
        """
        sig = inspect.signature(func)
        bound = sig.bind_partial(None, *args, **kwargs)
        bound.apply_defaults()
        arguments = dict(bound.arguments)
        arguments.pop("self", None)
        return arguments

    async def _execute_sql(
        self,
        session: Any,
        meta: dict[str, Any],
        params: dict[str, Any],
    ) -> Any:
        """Execute a raw-SQL-mode query.

        Dynamic sorting via ``PageRequest.sorts`` is rejected in this
        mode to prevent SQL injection.

        Raises:
            ValueError: If ``PageRequest.sorts`` is non-empty.
        """
        sql = meta.get("sql")
        unique = meta.get("unique", False)
        paged = meta.get("paged", False)

        page_req = _extract_page_request(params, paged)
        if page_req and page_req.sorts:
            raise ValueError(
                "Dynamic sorting via PageRequest is not supported in SQL mode. "
                "Please add ORDER BY clause to your SQL string directly."
            )

        if paged:
            return await _execute_paginated_query(session, sql, params, page_req)
        return await _execute_simple_query(session, sql, params, unique)

    async def _execute_expr(
        self,
        session: Any,
        meta: dict[str, Any],
        params: dict[str, Any],
        entity: Any,
    ) -> Any:
        """Execute an expression-mode query.

        Builds ``SELECT * FROM <table> WHERE <expr>`` using the entity's
        ``__tablename__``.  Dynamic sorting is supported and validated
        against the entity's column set.

        Raises:
            RuntimeError: If *entity* is ``None`` or lacks
                ``__tablename__``.
            ValueError: If a sort field is not a valid column.
        """
        expr = meta.get("expr")
        unique = meta.get("unique", False)
        paged = meta.get("paged", False)

        self._validate_entity(entity)
        base_sql = self._build_base_sql(entity, expr)

        page_req = _extract_page_request(params, paged)
        if page_req and page_req.sorts:
            valid_columns = {c.name for c in entity.__table__.columns}
            base_sql += _build_order_by_clause(page_req, valid_columns)

        if paged:
            return await _execute_paginated_query(session, base_sql, params, page_req)
        return await _execute_simple_query(session, base_sql, params, unique)

    def _validate_entity(self, entity: Any) -> None:
        """Ensure *entity* is set and has a ``__tablename__`` attribute.

        Raises:
            RuntimeError: ``"@query with expr requires @repository(entity=...)
                and an entity with __tablename__"``
        """
        if entity is None or not hasattr(entity, "__tablename__"):
            raise RuntimeError("@query with expr requires @repository(entity=...) and an entity with __tablename__")

    def _build_base_sql(self, entity: Any, expr: str | None) -> str:
        """Build the base ``SELECT`` query from the entity and optional expression.

        Args:
            entity: The SQLAlchemy model class (must have ``__tablename__``).
            expr: Optional WHERE-clause expression (e.g. ``"name = :name"``).

        Returns:
            A SQL string like ``"SELECT * FROM users WHERE name = :name"``.
        """
        table_name = entity.__tablename__
        base_sql = f"SELECT * FROM {table_name}"
        if expr:
            base_sql += f" WHERE {expr}"
        return base_sql