INNER CODE UNIT · Python

paginate_query

dialoguemd/fastapi-sqla · fastapi_sqla/async_pagination.py:26

async def paginate_query(
    query: Select,
    session: SqlaAsyncSession,
    total_items: int,
    offset: int,
    limit: int,
    *,
    scalars: bool = True,
) -> Page:
    total_pages = math.ceil(total_items / limit)
    page_number = math.floor(offset / limit + 1)
    query = query.offset(offset).limit(limit)
    result = await session.execute(query)
    data = iter(
        cast(Iterator, result.unique().scalars() if scalars else result.mappings())
    )
    return Page(
        data=data,  # type: ignore # Expected to be a list

View source record →