Custom Pagination Function

Let's create a custom pagination class that will allow us to paginate sqlalchemy queries.

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Step 1: create paginate function

First, we need to create a function that will paginate sqlalchemy queries. paginate is a common name for this function, but you can use any name you want.

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Step 2: define required function parameters

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Parameters:

• params - AbstractParams instance that contains pagination parameters.
• tranformer - function that will be used to transform items.
• additional_data - additional data that will be added to created page.

Step 3: verify that params

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Verify that params is a instance of limit-offset pagination params type.

Step 4: fetch total number of items

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Step 5: fetch items

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Step 6: transform items

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )

Step 7: create page and return it

from typing import Any, Optional

from sqlalchemy import func
from sqlalchemy.engine import Connection
from sqlalchemy.sql import Select

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractParams
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params


def paginate(
    conn: Connection,
    stmt: Select,
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    total = conn.scalar(stmt.with_only_columns(func.count()))
    q = stmt.offset(raw_params.offset).limit(raw_params.limit)
    items = conn.execute(q).all()

    t_items = apply_items_transformer(items, transformer)

    return create_page(
        t_items,
        total,
        params,
        **(additional_data or {}),
    )