Source code for fastapi_contrib.pagination
import asyncio
from fastapi import Query
from starlette.requests import Request
from fastapi_contrib.serializers.common import Serializer
[docs]class PaginationMeta(type):
def __new__(mcs, name, bases, namespace, *args, **kwargs):
cls = super(PaginationMeta, mcs).__new__(mcs, name, bases, namespace)
_cls__init__ = cls.__init__
def __init__(
self,
request: Request,
offset: int = Query(
default=cls.default_offset, ge=0, le=cls.max_offset
),
limit: int = Query(
default=cls.default_limit, ge=1, le=cls.max_limit
),
):
_cls__init__(self, request, offset, limit)
setattr(cls, "__init__", __init__)
return cls
[docs]class Pagination(metaclass=PaginationMeta):
"""
Query params parser and db collection paginator in one.
Use it as dependency in route, then invoke `paginate` with serializer:
.. code-block:: python
app = FastAPI()
class SomeSerializer(ModelSerializer):
class Meta:
model = SomeModel
@app.get("/")
async def list(pagination: Pagination = Depends()):
filter_kwargs = {}
return await pagination.paginate(
serializer_class=SomeSerializer, **filter_kwargs
)
Subclass this pagination to define custom
default & maximum values for offset & limit:
.. code-block:: python
class CustomPagination(Pagination):
default_offset = 90
default_limit = 1
max_offset = 100`
max_limit = 2000
:param request: starlette Request object
:param offset: query param of how many records to skip
:param limit: query param of how many records to show
"""
default_offset = 0
default_limit = 100
max_offset = None
max_limit = 1000
def __init__(
self,
request: Request,
offset: int = Query(default=default_offset, ge=0, le=max_offset),
limit: int = Query(default=default_limit, ge=1, le=max_limit),
):
self.request = request
self.offset = offset
self.limit = limit
self.model = None
self.count = None
self.list = []
[docs] async def get_count(self, **kwargs) -> int:
"""
Retrieves counts for query list, filtered by kwargs.
:param kwargs: filters that are proxied in db query
:return: number of found records
"""
self.count = await self.model.count(**kwargs)
return self.count
[docs] def get_next_url(self) -> str:
"""
Constructs `next` parameter in resulting JSON,
produces URL for next "page" of paginated results.
:return: URL for next "page" of paginated results.
"""
if self.offset + self.limit >= self.count:
return None
return str(
self.request.url.include_query_params(
limit=self.limit, offset=self.offset + self.limit
)
)
[docs] def get_previous_url(self) -> str:
"""
Constructs `previous` parameter in resulting JSON,
produces URL for previous "page" of paginated results.
:return: URL for previous "page" of paginated results.
"""
if self.offset <= 0:
return None
if self.offset - self.limit <= 0:
return str(self.request.url.remove_query_params(keys=["offset"]))
return str(
self.request.url.include_query_params(
limit=self.limit, offset=self.offset - self.limit
)
)
[docs] async def get_list(self, _sort=None, **kwargs) -> list:
"""
Retrieves actual list of records. It comes raw, which means
it retrieves dict from DB, instead of making conversion
for every object in list into Model.
:param kwargs: filters that are proxied in db query
:return: list of dicts from DB, filtered by kwargs
"""
self.list = await self.model.list(
_limit=self.limit,
_offset=self.offset,
_sort=_sort,
raw=True,
**kwargs
)
return self.list
[docs] async def paginate(
self, serializer_class: Serializer, _sort=None, **kwargs
) -> dict:
"""
Actual pagination function, takes serializer class,
filter options as kwargs and returns dict with the following fields:
* count - counts for query list, filtered by kwargs
* next - URL for next "page" of paginated results
* previous - URL for previous "page" of paginated results
* result - actual list of records (dicts)
:param serializer_class: needed to get Model & sanitize list from DB
:param kwargs: filters that are proxied in db query
:return: dict that should be returned as a response
"""
self.model = serializer_class.Meta.model
count, _list = await asyncio.gather(
self.get_count(**kwargs), self.get_list(_sort=_sort, **kwargs)
)
# TODO: think about naming and separation of concerns
_list = serializer_class.sanitize_list(_list)
return {
"count": count,
"next": self.get_next_url(),
"previous": self.get_previous_url(),
"result": _list,
}