Standard Pagination

ZodiacCore provides a comprehensive pagination system that standardizes how your API handles list-based data. It includes request parameters, response models, and professional repository methods that automate pagination logic.

1. Request Parameters

The PageParams model handles typical pagination query strings (?page=1&size=20). Use SortParams or PageSortParams when an endpoint also supports repeated multi-column sort parameters such as ?sort=name:asc&sort=created_at:desc.

from typing import Annotated

from fastapi import Query
from zodiac_core.pagination import PageParams
from zodiac_core.routing import APIRouter

router = APIRouter()


@router.get("/items")
async def list_items(
    params: Annotated[PageParams, Query()],
):
    # Automatically validated:
    # params.page defaults to 1 (min 1)
    # params.size defaults to 20 (max 100)
    ...

Query Parameter Models

FastAPI officially documents Pydantic query parameter models with Query(). PageParams follows that pattern: FastAPI extracts page and size from the query string and validates them against the model.

---

2. Multi-Column Sorting

Use PageSortParams for paginated list APIs that support sorting:

from typing import Annotated

from fastapi import Query
from zodiac_core.pagination import PageSortParams


@router.get("/items")
async def list_items(
    params: Annotated[PageSortParams, Query()],
):
    # /items?page=1&size=20&sort=name:asc&sort=created_at:desc
    # params.sort == ["name:asc", "created_at:desc"]
    ...

Use SortParams directly for endpoints that need sorting without pagination. Sort directions are limited to asc and desc; when omitted, the direction defaults to asc.

If you do not use ZodiacCore repository helpers, consume params.sort_pairs directly:

for field, direction in params.sort_pairs:
    ...

For SQL repositories, define a reusable SortSpec once on the repository:

from sqlalchemy import select
from zodiac_core.pagination import PageSortParams, SortSpec


class ItemRepository(BaseSQLRepository):
    sort_spec = SortSpec(
        columns={
            "name": ItemModel.name,
            "created_at": ItemModel.created_at,
        },
        default=["created_at:desc", "name:asc"],
    )

    async def list_items(self, params: PageSortParams):
        return await self.paginate_query(select(ItemModel), params)

The public query field is a list of strings, so OpenAPI documents sort as repeated string parameters:

?sort=name:asc&sort=created_at:desc

paginate_query() parses those strings internally and applies the repository SortSpec. Unknown sort fields raise BadRequestException; this keeps public API field names explicit and avoids sorting by arbitrary database columns.

For one-off sorting configuration, pass an explicit SortSpec:

return await self.paginate_query(
    select(ItemModel),
    params,
    sort_spec=SortSpec(
        columns={
            "name": ItemModel.name,
            "created_at": ItemModel.created_at,
        }
    ),
)

---

3. Standard Paged Response

The PagedResponse[T] is a generic model that wraps your data items along with metadata.

The Response Structure

{
  "code": 0,
  "message": "Success",
  "data": {
    "items": [...],
    "total": 100,
    "page": 1,
    "size": 20
  }
}

Building the Response

Use the .create() factory method to easily build the response from your query results and the input PageParams.

from zodiac_core.pagination import PagedResponse

return PagedResponse.create(
    items=items,
    total=total_count,
    params=page_params
)

---

4. Professional Pagination with BaseSQLRepository

For database queries, BaseSQLRepository provides methods that automate pagination and sorting:

paginate_query() - Recommended for Most Cases

This is the convenience method that automatically manages the database session. Use this in your repository methods:

from sqlalchemy import select
from zodiac_core.db.repository import BaseSQLRepository
from zodiac_core.pagination import PagedResponse, PageSortParams, SortSpec


class ItemRepository(BaseSQLRepository):
    sort_spec = SortSpec(
        columns={
            "id": ItemModel.id,
            "name": ItemModel.name,
            "created_at": ItemModel.created_at,
        },
        default=["id:asc"],
    )

    async def list_items(self, params: PageSortParams) -> PagedResponse[ItemModel]:
        """List items with pagination."""
        stmt = select(ItemModel)
        return await self.paginate_query(stmt, params)

What it does:

• ✅ Automatically manages database session
• ✅ Optionally applies multi-column sorting with repository-level or per-call SortSpec
• ✅ Calculates total count (handles complex queries with joins/groups)
• ✅ Applies limit/offset
• ✅ Packages results into PagedResponse

When to use: - Most repository methods that need pagination - Simple queries that don't require custom session management

paginate() - For Advanced Use Cases

This method requires you to manage the session yourself. Use this when you need more control:

async def list_items_with_custom_logic(self, params: PageParams) -> PagedResponse[ItemModel]:
    """Example with custom session management."""
    async with self.session() as session:
        # You can add custom logic here (e.g., filtering, joins)
        stmt = select(ItemModel).where(ItemModel.status == "active")
        stmt = stmt.order_by(ItemModel.created_at.desc())

        return await self.paginate(session, stmt, params)

What it does:

• ✅ Calculates total count (handles complex queries)
• ✅ Applies limit/offset
• ✅ Packages results into PagedResponse
• ⚠️ Requires you to provide an active session

When to use:

• When you need custom session management
• When you want to perform multiple operations in a single transaction
• When you need to add complex query logic before pagination

How Count Calculation Works

Both methods handle complex queries correctly:

• Simple queries: SELECT COUNT(*) FROM (SELECT ...)
• Queries with joins: Automatically wraps in subquery
• Queries with GROUP BY: Handles correctly
• Queries with ORDER BY: Removed from count query (as expected)

The implementation removes limit/offset before counting and safely wraps complex queries in subqueries.

Transformation Support

Both methods support optional transformation to Pydantic models:

from app.api.schemas.item_schema import ItemSchema

# Transform DB models to response schemas
return await self.paginate_query(stmt, params, transformer=ItemSchema)

---

5. Complete Example

Here's a complete example showing the full flow:

Repository:

from zodiac_core.pagination import PagedResponse, PageSortParams, SortSpec


class ItemRepository(BaseSQLRepository):
    sort_spec = SortSpec(columns={"id": ItemModel.id, "name": ItemModel.name}, default=["id:asc"])

    async def list_items(self, params: PageSortParams) -> PagedResponse[ItemModel]:
        return await self.paginate_query(select(ItemModel), params)

Service:

from zodiac_core.pagination import PagedResponse, PageSortParams


class ItemService:
    def __init__(self, item_repo: ItemRepository) -> None:
        self.item_repo = item_repo

    async def list_items(self, page_params: PageSortParams) -> PagedResponse[ItemModel]:
        return await self.item_repo.list_items(page_params)

Router:

from typing import Annotated

from dependency_injector.wiring import Provide, inject
from fastapi import Depends, Query
from zodiac_core.pagination import PagedResponse, PageSortParams


@router.get("", response_model=PagedResponse[ItemSchema])
@inject
async def list_items(
    page_params: Annotated[PageSortParams, Query()],
    service: Annotated[ItemService, Depends(Provide[Container.item_service])],
):
    return await service.list_items(page_params)

No manual calculations needed! The paginate_query method handles everything.

---

6. API Reference

Pagination Models

PageParams

Bases: BaseModel

Standard pagination query parameters.

Usage

from typing import Annotated
from fastapi import Query
from zodiac_core.pagination import PageParams

@app.get("/users")
def list_users(page_params: Annotated[PageParams, Query()]):
    skip = (page_params.page - 1) * page_params.size
    limit = page_params.size
    ...

Source code in zodiac_core/pagination.py

class PageParams(BaseModel):
    """
    Standard pagination query parameters.

    Usage:
        ```python
        from typing import Annotated
        from fastapi import Query
        from zodiac_core.pagination import PageParams

        @app.get("/users")
        def list_users(page_params: Annotated[PageParams, Query()]):
            skip = (page_params.page - 1) * page_params.size
            limit = page_params.size
            ...
        ```
    """

    page: int = Field(1, ge=1, description="Page number (1-based)")
    size: int = Field(20, ge=1, le=100, description="Page size")

SortParams

Bases: BaseModel

Standard multi-column sorting query parameters.

FastAPI maps repeated sort query parameters into the list: ?sort=name:asc&sort=created_at:desc.

Source code in zodiac_core/pagination.py

class SortParams(BaseModel):
    """
    Standard multi-column sorting query parameters.

    FastAPI maps repeated `sort` query parameters into the list:
    `?sort=name:asc&sort=created_at:desc`.
    """

    sort: List[str] = Field(
        default_factory=list,
        description='Sort expressions in "field:direction" format, for example "name:asc".',
    )

    @field_validator("sort")
    @classmethod
    def validate_sort_expressions(cls, values: List[str]) -> List[str]:
        _parse_sort_items(values)
        return values

    @property
    def sort_pairs(self) -> List[SortPair]:
        """
        Parsed sort fields for consumers that do not use repository helpers.
        """
        return [(item.field, item.direction) for item in _parse_sort_items(self.sort)]

sort_pairs property

Parsed sort fields for consumers that do not use repository helpers.

SortSpec dataclass

Reusable sorting configuration for repository list queries.

columns is an explicit whitelist from public API field names to SQLAlchemy column/expression objects. default is used when the request does not include a sort parameter.

Example

class ItemRepository(BaseSQLRepository):
    sort_spec = SortSpec(
        columns={
            "name": ItemModel.name,
            "created_at": ItemModel.created_at,
        },
        default=["created_at:desc", "name:asc"],
    )

Source code in zodiac_core/pagination.py

@dataclass(frozen=True)
class SortSpec:
    """
    Reusable sorting configuration for repository list queries.

    ``columns`` is an explicit whitelist from public API field names to
    SQLAlchemy column/expression objects. ``default`` is used when the request
    does not include a sort parameter.

    Example:
        ```python
        class ItemRepository(BaseSQLRepository):
            sort_spec = SortSpec(
                columns={
                    "name": ItemModel.name,
                    "created_at": ItemModel.created_at,
                },
                default=["created_at:desc", "name:asc"],
            )
        ```
    """

    columns: Mapping[str, Any]
    default: Sequence[SortExpression] = dataclass_field(default_factory=tuple)

    def __post_init__(self) -> None:
        columns = MappingProxyType(dict(self.columns))
        default_pairs = tuple(pair for expression in self.default for pair in _coerce_sort_expression(expression))
        unknown_defaults = [field for field, _ in default_pairs if field not in columns]
        if unknown_defaults:
            unknown = ", ".join(sorted(set(unknown_defaults)))
            raise ValueError(f"Default sort field is not configured in SortSpec.columns: {unknown}")

        object.__setattr__(self, "columns", columns)
        object.__setattr__(self, "_default_pairs", default_pairs)

    @property
    def default_pairs(self) -> List[SortPair]:
        """Parsed default sort fields."""
        return list(self._default_pairs)

    def pairs_for(self, sort_params: "SortParams | None") -> List[SortPair]:
        """Return request sort pairs when present; otherwise return defaults."""
        if sort_params is not None and sort_params.sort_pairs:
            return sort_params.sort_pairs
        return self.default_pairs

default_pairs property

Parsed default sort fields.

pairs_for(sort_params)

Return request sort pairs when present; otherwise return defaults.

Source code in zodiac_core/pagination.py

def pairs_for(self, sort_params: "SortParams | None") -> List[SortPair]:
    """Return request sort pairs when present; otherwise return defaults."""
    if sort_params is not None and sort_params.sort_pairs:
        return sort_params.sort_pairs
    return self.default_pairs

PageSortParams

Bases: PageParams, SortParams

Combined pagination and sorting query parameters.

Source code in zodiac_core/pagination.py

class PageSortParams(PageParams, SortParams):
    """
    Combined pagination and sorting query parameters.
    """

PagedResponse

Bases: BaseModel, Generic[T]

Standard generic paginated response model.

Usage

from typing import Annotated
from fastapi import Query
from zodiac_core.pagination import PagedResponse, PageParams

@app.get("/users", response_model=PagedResponse[UserSchema])
def list_users(page_params: Annotated[PageParams, Query()]):
    users, total_count = db.find_users(...)
    return PagedResponse.create(users, total_count, page_params)

Source code in zodiac_core/pagination.py

class PagedResponse(BaseModel, Generic[T]):
    """
    Standard generic paginated response model.

    Usage:
        ```python
        from typing import Annotated
        from fastapi import Query
        from zodiac_core.pagination import PagedResponse, PageParams

        @app.get("/users", response_model=PagedResponse[UserSchema])
        def list_users(page_params: Annotated[PageParams, Query()]):
            users, total_count = db.find_users(...)
            return PagedResponse.create(users, total_count, page_params)
        ```
    """

    model_config = ConfigDict(populate_by_name=True)

    items: List[T] = Field(description="List of items for the current page")
    total: int = Field(description="Total number of items")
    page: int = Field(description="Current page number")
    size: int = Field(description="Current page size")

    @classmethod
    def create(
        cls,
        items: List[T],
        total: int,
        params: PageParams,
    ) -> "PagedResponse[T]":
        """
        Factory method to create a PagedResponse from items, total count, and PageParams.

        Args:
            items: The list of data objects (Pydantic models or dicts).
            total: The total number of records in the database matching the query.
            params: The PageParams object from the request.
        """
        return cls(
            items=items,
            total=total,
            page=params.page,
            size=params.size,
        )

create(items, total, params) classmethod

Factory method to create a PagedResponse from items, total count, and PageParams.

Parameters:

Name	Type	Description	Default
items	List[T]	The list of data objects (Pydantic models or dicts).	required
total	int	The total number of records in the database matching the query.	required
params	PageParams	The PageParams object from the request.	required

Source code in zodiac_core/pagination.py

@classmethod
def create(
    cls,
    items: List[T],
    total: int,
    params: PageParams,
) -> "PagedResponse[T]":
    """
    Factory method to create a PagedResponse from items, total count, and PageParams.

    Args:
        items: The list of data objects (Pydantic models or dicts).
        total: The total number of records in the database matching the query.
        params: The PageParams object from the request.
    """
    return cls(
        items=items,
        total=total,
        page=params.page,
        size=params.size,
    )

Repository Methods

Standard base class for SQL-based repositories.

Supports multiple database instances via db_name and provides professional utilities for common operations like pagination.

Source code in zodiac_core/db/repository.py

class BaseSQLRepository:
    """
    Standard base class for SQL-based repositories.

    Supports multiple database instances via `db_name` and provides
    professional utilities for common operations like pagination.
    """

    sort_spec: SortSpec | None = None

    def __init__(
        self,
        session_factory: Optional[async_sessionmaker[AsyncSession]] = None,
        db_name: str = DEFAULT_DB_NAME,
        options: Optional[Any] = None,
    ) -> None:
        """
        Initialize the repository.

        Args:
            session_factory: Optional custom session factory. If provided, db_name is ignored.
            db_name: The name of the database engine registered in db.setup(). Defaults to DEFAULT_DB_NAME ("default").
            options: Optional configuration/options for the repository.
        """
        self._session_factory = session_factory
        self.db_name = db_name
        self.options = options

    @asynccontextmanager
    async def session(self) -> AsyncIterator[AsyncSession]:
        """
        Async context manager for obtaining a database session.
        Uses the injected factory or resolves one from the global 'db' via 'db_name'.

        Note:
            This context manager does NOT auto-commit. You must explicitly call
            `await session.commit()` to persist changes to the database.
        """
        factory = self._session_factory or db.get_factory(self.db_name)
        async with manage_session(factory) as session:
            yield session

    async def paginate(
        self,
        session: AsyncSession,
        statement: Any,
        params: PageParams,
        transformer: Optional[Type[T]] = None,
        *,
        sort_spec: SortSpec | None = None,
    ) -> PagedResponse[T]:
        """
        Execute a paginated query with automatic count and paging.

        Performs:
        1. Optional multi-column sorting using public field mappings.
        2. Automatic total count query using the provided statement.
        3. Automatic limit/offset application.
        4. Packaging results into a standardized PagedResponse.

        Args:
            session: The active AsyncSession.
            statement: The SQLAlchemy select statement (without limit/offset).
            params: Standard PageParams (page, size).
            transformer: Optional Pydantic model to transform DB objects into.
            sort_spec: Optional reusable sort configuration. When omitted, the
                repository class-level ``sort_spec`` is used.

        Example:
            ```python
            async with self.session() as session:
                stmt = select(UserModel).order_by(UserModel.created_at.desc())
                return await self.paginate(session, stmt, params)
            ```
        """
        effective_sort_spec = sort_spec or self.sort_spec

        if effective_sort_spec is not None:
            sort_params = params if isinstance(params, SortParams) else None
            statement = self.apply_sorting(statement, sort_params, sort_spec=effective_sort_spec)

        # 1. Execute Count Query
        # Remove limit/offset/order_by (if any) for count query, then wrap in subquery.
        # Wrapping in subquery handles complex queries (joins, groups).
        count_base = statement.limit(None).offset(None).order_by(None)
        count_stmt = select(func.count()).select_from(count_base.subquery())
        total = (await session.execute(count_stmt)).scalar() or 0

        # 2. Execute Paged Query
        skip = (params.page - 1) * params.size
        paged_stmt = statement.offset(skip).limit(params.size)
        result = await session.execute(paged_stmt)
        items = result.scalars().all()

        # 3. Optional Transformation
        if transformer:
            items = [transformer.model_validate(item) for item in items]

        return PagedResponse.create(items=list(items), total=total, params=params)

    def apply_sorting(
        self,
        statement: Any,
        sort_params: SortParams | None = None,
        *,
        sort_spec: SortSpec | None = None,
    ) -> Any:
        """
        Apply validated multi-column sorting to a SQLAlchemy statement.
        Existing ORDER BY clauses are replaced when sort fields are present.

        Args:
            statement: The SQLAlchemy select statement to sort.
            sort_params: Standard SortParams or PageSortParams. When omitted,
                only the configured default sort is applied.
            sort_spec: Optional reusable sort configuration. When omitted, the
                repository class-level ``sort_spec`` is used.

        Example:
            ```python
            stmt = self.apply_sorting(
                select(UserModel),
                params,
                sort_spec=SortSpec(
                    columns={
                        "name": UserModel.name,
                        "created_at": UserModel.created_at,
                    }
                ),
            )
            ```
        """
        if sort_params is not None and not isinstance(sort_params, SortParams):
            raise TypeError("sort_params must be SortParams or PageSortParams")

        effective_sort_spec = sort_spec or self.sort_spec

        if effective_sort_spec is None:
            return statement

        sort_pairs = effective_sort_spec.pairs_for(sort_params)
        if not sort_pairs:
            return statement

        order_by_clauses = []
        for field, direction in sort_pairs:
            column = effective_sort_spec.columns.get(field)
            if column is None:
                supported_fields = sorted(effective_sort_spec.columns)
                raise BadRequestException(
                    message=(
                        f"Unsupported sort field '{field}'. Supported sort fields: {', '.join(supported_fields)}."
                    ),
                    data={
                        "field": field,
                        "supported_fields": supported_fields,
                    },
                )
            order_by_clauses.append(column.asc() if direction == "asc" else column.desc())

        return statement.order_by(None).order_by(*order_by_clauses)

    async def paginate_query(
        self,
        statement: Any,
        params: PageParams,
        transformer: Optional[Type[T]] = None,
        *,
        sort_spec: SortSpec | None = None,
    ) -> PagedResponse[T]:
        """
        Convenience method that automatically manages session for pagination.

        This is a wrapper around `paginate()` that handles session management,
        making it easier to use in repository methods.

        Args:
            statement: The SQLAlchemy select statement (without limit/offset).
            params: Standard PageParams (page, size).
            transformer: Optional Pydantic model to transform DB objects into.
            sort_spec: Optional reusable sort configuration. When omitted, the
                repository class-level ``sort_spec`` is used.

        Example:
            ```python
            async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
                stmt = select(ItemModel).order_by(ItemModel.id)
                return await self.paginate_query(stmt, params)
            ```
        """
        async with self.session() as session:
            return await self.paginate(
                session,
                statement,
                params,
                transformer,
                sort_spec=sort_spec,
            )

apply_sorting(statement, sort_params=None, *, sort_spec=None)

Apply validated multi-column sorting to a SQLAlchemy statement. Existing ORDER BY clauses are replaced when sort fields are present.

Parameters:

Name	Type	Description	Default
statement	Any	The SQLAlchemy select statement to sort.	required
sort_params	SortParams | None	Standard SortParams or PageSortParams. When omitted, only the configured default sort is applied.	None
sort_spec	SortSpec | None	Optional reusable sort configuration. When omitted, the repository class-level sort_spec is used.	None

Example

stmt = self.apply_sorting(
    select(UserModel),
    params,
    sort_spec=SortSpec(
        columns={
            "name": UserModel.name,
            "created_at": UserModel.created_at,
        }
    ),
)

Source code in zodiac_core/db/repository.py

def apply_sorting(
    self,
    statement: Any,
    sort_params: SortParams | None = None,
    *,
    sort_spec: SortSpec | None = None,
) -> Any:
    """
    Apply validated multi-column sorting to a SQLAlchemy statement.
    Existing ORDER BY clauses are replaced when sort fields are present.

    Args:
        statement: The SQLAlchemy select statement to sort.
        sort_params: Standard SortParams or PageSortParams. When omitted,
            only the configured default sort is applied.
        sort_spec: Optional reusable sort configuration. When omitted, the
            repository class-level ``sort_spec`` is used.

    Example:
        ```python
        stmt = self.apply_sorting(
            select(UserModel),
            params,
            sort_spec=SortSpec(
                columns={
                    "name": UserModel.name,
                    "created_at": UserModel.created_at,
                }
            ),
        )
        ```
    """
    if sort_params is not None and not isinstance(sort_params, SortParams):
        raise TypeError("sort_params must be SortParams or PageSortParams")

    effective_sort_spec = sort_spec or self.sort_spec

    if effective_sort_spec is None:
        return statement

    sort_pairs = effective_sort_spec.pairs_for(sort_params)
    if not sort_pairs:
        return statement

    order_by_clauses = []
    for field, direction in sort_pairs:
        column = effective_sort_spec.columns.get(field)
        if column is None:
            supported_fields = sorted(effective_sort_spec.columns)
            raise BadRequestException(
                message=(
                    f"Unsupported sort field '{field}'. Supported sort fields: {', '.join(supported_fields)}."
                ),
                data={
                    "field": field,
                    "supported_fields": supported_fields,
                },
            )
        order_by_clauses.append(column.asc() if direction == "asc" else column.desc())

    return statement.order_by(None).order_by(*order_by_clauses)

paginate(session, statement, params, transformer=None, *, sort_spec=None) async

Execute a paginated query with automatic count and paging.

Performs: 1. Optional multi-column sorting using public field mappings. 2. Automatic total count query using the provided statement. 3. Automatic limit/offset application. 4. Packaging results into a standardized PagedResponse.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The active AsyncSession.	required
statement	Any	The SQLAlchemy select statement (without limit/offset).	required
params	PageParams	Standard PageParams (page, size).	required
transformer	Optional[Type[T]]	Optional Pydantic model to transform DB objects into.	None
sort_spec	SortSpec | None	Optional reusable sort configuration. When omitted, the repository class-level sort_spec is used.	None

Example

async with self.session() as session:
    stmt = select(UserModel).order_by(UserModel.created_at.desc())
    return await self.paginate(session, stmt, params)

Source code in zodiac_core/db/repository.py

async def paginate(
    self,
    session: AsyncSession,
    statement: Any,
    params: PageParams,
    transformer: Optional[Type[T]] = None,
    *,
    sort_spec: SortSpec | None = None,
) -> PagedResponse[T]:
    """
    Execute a paginated query with automatic count and paging.

    Performs:
    1. Optional multi-column sorting using public field mappings.
    2. Automatic total count query using the provided statement.
    3. Automatic limit/offset application.
    4. Packaging results into a standardized PagedResponse.

    Args:
        session: The active AsyncSession.
        statement: The SQLAlchemy select statement (without limit/offset).
        params: Standard PageParams (page, size).
        transformer: Optional Pydantic model to transform DB objects into.
        sort_spec: Optional reusable sort configuration. When omitted, the
            repository class-level ``sort_spec`` is used.

    Example:
        ```python
        async with self.session() as session:
            stmt = select(UserModel).order_by(UserModel.created_at.desc())
            return await self.paginate(session, stmt, params)
        ```
    """
    effective_sort_spec = sort_spec or self.sort_spec

    if effective_sort_spec is not None:
        sort_params = params if isinstance(params, SortParams) else None
        statement = self.apply_sorting(statement, sort_params, sort_spec=effective_sort_spec)

    # 1. Execute Count Query
    # Remove limit/offset/order_by (if any) for count query, then wrap in subquery.
    # Wrapping in subquery handles complex queries (joins, groups).
    count_base = statement.limit(None).offset(None).order_by(None)
    count_stmt = select(func.count()).select_from(count_base.subquery())
    total = (await session.execute(count_stmt)).scalar() or 0

    # 2. Execute Paged Query
    skip = (params.page - 1) * params.size
    paged_stmt = statement.offset(skip).limit(params.size)
    result = await session.execute(paged_stmt)
    items = result.scalars().all()

    # 3. Optional Transformation
    if transformer:
        items = [transformer.model_validate(item) for item in items]

    return PagedResponse.create(items=list(items), total=total, params=params)

paginate_query(statement, params, transformer=None, *, sort_spec=None) async

Convenience method that automatically manages session for pagination.

This is a wrapper around paginate() that handles session management, making it easier to use in repository methods.

Parameters:

Name	Type	Description	Default
statement	Any	The SQLAlchemy select statement (without limit/offset).	required
params	PageParams	Standard PageParams (page, size).	required
transformer	Optional[Type[T]]	Optional Pydantic model to transform DB objects into.	None
sort_spec	SortSpec | None	Optional reusable sort configuration. When omitted, the repository class-level sort_spec is used.	None

Example

async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
    stmt = select(ItemModel).order_by(ItemModel.id)
    return await self.paginate_query(stmt, params)

Source code in zodiac_core/db/repository.py

async def paginate_query(
    self,
    statement: Any,
    params: PageParams,
    transformer: Optional[Type[T]] = None,
    *,
    sort_spec: SortSpec | None = None,
) -> PagedResponse[T]:
    """
    Convenience method that automatically manages session for pagination.

    This is a wrapper around `paginate()` that handles session management,
    making it easier to use in repository methods.

    Args:
        statement: The SQLAlchemy select statement (without limit/offset).
        params: Standard PageParams (page, size).
        transformer: Optional Pydantic model to transform DB objects into.
        sort_spec: Optional reusable sort configuration. When omitted, the
            repository class-level ``sort_spec`` is used.

    Example:
        ```python
        async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
            stmt = select(ItemModel).order_by(ItemModel.id)
            return await self.paginate_query(stmt, params)
        ```
    """
    async with self.session() as session:
        return await self.paginate(
            session,
            statement,
            params,
            transformer,
            sort_spec=sort_spec,
        )