Skip to content

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).

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

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

Using Depends() vs Query()

For Pydantic models like PageParams, use Depends() instead of Query(). FastAPI will automatically extract query parameters and validate them against the model.

2. 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
)

3. Professional Pagination with BaseSQLRepository

For database queries, BaseSQLRepository provides two methods that automate pagination:

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, PageParams

class ItemRepository(BaseSQLRepository):
    async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
        """List items with pagination."""
        stmt = select(ItemModel).order_by(ItemModel.id)
        return await self.paginate_query(stmt, params)

What it does:

  • ✅ Automatically manages database session
  • ✅ 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)

4. Complete Example

Here's a complete example showing the full flow:

Repository: 

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

Service: 

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

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

Router: 

@router.get("", response_model=PagedResponse[ItemSchema])
@inject
async def list_items(
    page_params: Annotated[PageParams, Depends()],
    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.

5. 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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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")

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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@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 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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.
    """

    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,
    ) -> PagedResponse[T]:
        """
        Execute a paginated query with automatic count and paging.

        Performs:
        1. Automatic total count query using the provided statement.
        2. Automatic limit/offset application.
        3. 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.

        Example:
            ```python
            async with self.session() as session:
                stmt = select(UserModel).order_by(UserModel.created_at.desc())
                return await self.paginate(session, stmt, params)
            ```
        """
        # 1. Execute Count Query
        # Remove limit/offset (if any) for count query, then wrap in subquery
        # subquery() handles order_by correctly, and wrapping in subquery handles complex queries (joins, groups)
        count_base = statement.limit(None).offset(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)

    async def paginate_query(
        self,
        statement: Any,
        params: PageParams,
        transformer: Optional[Type[T]] = 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.

        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)

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

Execute a paginated query with automatic count and paging.

Performs: 1. Automatic total count query using the provided statement. 2. Automatic limit/offset application. 3. 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
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def paginate(
    self,
    session: AsyncSession,
    statement: Any,
    params: PageParams,
    transformer: Optional[Type[T]] = None,
) -> PagedResponse[T]:
    """
    Execute a paginated query with automatic count and paging.

    Performs:
    1. Automatic total count query using the provided statement.
    2. Automatic limit/offset application.
    3. 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.

    Example:
        ```python
        async with self.session() as session:
            stmt = select(UserModel).order_by(UserModel.created_at.desc())
            return await self.paginate(session, stmt, params)
        ```
    """
    # 1. Execute Count Query
    # Remove limit/offset (if any) for count query, then wrap in subquery
    # subquery() handles order_by correctly, and wrapping in subquery handles complex queries (joins, groups)
    count_base = statement.limit(None).offset(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) 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
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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def paginate_query(
    self,
    statement: Any,
    params: PageParams,
    transformer: Optional[Type[T]] = 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.

    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)