I have a simple REST api which is a books store created with FastAPI and mongo db as the backend (I have used Motor as the library instead of Pymongo). I have a GET endpoint to get all the books in the database which also supports query strings (For example : A user can search for books with a single author or with a genre type etc).
Below are the corresponding codes for this endpoint :
routers.py
@router.get("/books", response_model=List[models.AllBooksResponse])
async def get_the_list_of_all_books(
authors: Optional[str] = None,
genres: Optional[str] = None,
published_year: Optional[str] = None,
) -> List[Dict[str, Any]]:
if authors is None and genres is None and published_year is None:
all_books = [book for book in await mongo.BACKEND.get_all_books()]
else:
all_books = [
book
for book in await mongo.BACKEND.get_all_books(
authors=authors.strip('"').split(",") if authors is not None else None,
genres=genres.strip('"').split(",") if genres is not None else None,
published_year=datetime.strptime(published_year, "%Y")
if published_year is not None
else None,
)
]
return all_books
The corresponding model :
class AllBooksResponse(BaseModel):
name: str
author: str
link: Optional[str] = None
def __init__(self, name, author, **data):
super().__init__(
name=name, author=author, link=f"{base_uri()}book/{data['book_id']}"
)
And the backend function for getting the data:
class MongoBackend:
def __init__(self, uri: str) -> None:
self._client = motor.motor_asyncio.AsyncIOMotorClient(uri)
async def get_all_books(
self,
authors: Optional[List[str]] = None,
genres: Optional[List[str]] = None,
published_year: Optional[datetime] = None,
) -> List[Dict[str, Any]]:
find_condition = {}
if authors is not None:
find_condition["author"] = {"$in": authors}
if genres is not None:
find_condition["genres"] = {"$in": genres}
if published_year is not None:
find_condition["published_year"] = published_year
cursor = self._client[DB][BOOKS_COLLECTION].find(find_condition, {"_id": 0})
return [doc async for doc in cursor]
Now i want to implement pagination for this endpoint . Here i have a few questions :
- Is it good to do pagination on database level or application level ?
- Do we have some out of the box libraries which can help me do that in fastapi ? I checked the documentation for https://pypi.org/project/fastapi-pagination/ , but this seems to be more targeted towards SQL databases
- I also checked out this link : https://www.codementor.io/@arpitbhayani/fast-and-efficient-pagination-in-mongodb-9095flbqr which talks about different ways of doing this in Mongo db but i think only the first option(using
limitandskip) would work for me, because i want to also make it work when i am using other filter parameters (for example for author and genre) and there is no way i can know the ObjectId's unless i make the first query to get the data and then i want to do pagination.
But the issue is everywhere i see using limit and skip is discouraged.
Can someone please let me know what are the best practices here and can something apply to my requirement and use case?
Many thanks in advance.