Merge pull request #36 from igorbenav/paginated-return

Helper functions for paginated responses, new module created for pagination

Author: igorbenav

README.md

@@ -96,6 +96,7 @@
     5. [Alembic Migrations](#55-alembic-migrations)
     6. [CRUD](#56-crud)
     7. [Routes](#57-routes)
+        1. [Paginated Responses](#571-paginated-responses)
     8. [Caching](#58-caching)
     9. [More Advanced Caching](#59-more-advanced-caching)
     10. [ARQ Job Queues](#510-arq-job-queues)
@@ -412,6 +413,7 @@ First, you may want to take a look at the project structure and understand what
     │   │   ├── __init__.py
     │   │   ├── dependencies.py       # Defines dependencies that can be reused across the API endpoints.
     │   │   ├── exceptions.py         # Contains custom exceptions for the API.
+    │   │   ├── paginated.py          # Provides utilities for paginated responses in APIs
     │   │   │
     │   │   └── v1                    # Version 1 of the API.
     │   │       ├── __init__.py
@@ -690,10 +692,15 @@ from app.core.database import async_get_db
 
 router = fastapi.APIRouter(tags=["entities"])
 
-@router.get("/entities", response_model=List[EntityRead])
-async def read_entities(db: Annotated[AsyncSession, Depends(async_get_db)]):
-  entities = await crud_entities.get_multi(db=db)
-    return entities
+@router.get("/entities/{id}", response_model=List[EntityRead])
+async def read_entities(
+  request: Request,
+  id: int,
+  db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+  entity = await crud_entities.get(db=db, id=id)  
+  
+  return entity
 
 ...
 ```
@@ -708,6 +715,71 @@ router = APIRouter(prefix="/v1") # this should be there already
 router.include_router(entity_router)
 ```
 
+#### 5.7.1 Paginated Responses
+With the `get_multi` method we get a python `dict` with full suport for pagination:
+```javascript
+{
+  "data": [
+    {
+      "id": 4,
+      "name": "User Userson",
+      "username": "userson4",
+      "email": "[email protected]",
+      "profile_image_url": "https://profileimageurl.com"
+    },
+    {
+      "id": 5,
+      "name": "User Userson",
+      "username": "userson5",
+      "email": "[email protected]",
+      "profile_image_url": "https://profileimageurl.com"
+    }
+  ],
+  "total_count": 2,
+  "has_more": false,
+  "page": 1,
+  "items_per_page": 10
+} 
+```
+
+And in the endpoint, we can import from `app/api/paginated` the following functions and Pydantic Schema:
+```python
+from app.api.paginated import (
+  PaginatedListResponse, # What you'll use as a response_model to validate
+  paginated_response,    # Creates a paginated response based on the parameters
+  compute_offset         # Calculate the offset for pagination ((page - 1) * items_per_page)
+)
+```
+
+Then let's create the endpoint:
+```python
+import fastapi
+
+from app.schemas.entity imoport EntityRead
+...
+
+@router.get("/entities", response_model=PaginatedListResponse[EntityRead])
+async def read_entities(
+    request: Request, 
+    db: Annotated[AsyncSession, Depends(async_get_db)],
+    page: int = 1,
+    items_per_page: int = 10
+):
+    entities_data = await crud_entity.get_multi(
+        db=db,
+        offset=compute_offset(page, items_per_page),
+        limit=items_per_page,
+        schema_to_select=UserRead, 
+        is_deleted=False
+    )
+    
+    return paginated_response(
+        crud_data=entities_data, 
+        page=page,
+        items_per_page=items_per_page
+    )
+```
+
 ### 5.8 Caching
 The `cache` decorator allows you to cache the results of FastAPI endpoint functions, enhancing response times and reducing the load on your application by storing and retrieving data in a cache.
 

src/app/api/paginated.py

@@ -0,0 +1,79 @@
+from typing import TypeVar, Generic, List
+
+from pydantic import BaseModel
+
+SchemaType = TypeVar("SchemaType", bound=BaseModel)
+
+class ListResponse(BaseModel, Generic[SchemaType]):
+    data: List[SchemaType]
+
+
+class PaginatedListResponse(ListResponse[SchemaType]):
+    total_count: int
+    has_more: bool
+    page: int | None = None
+    items_per_page: int | None = None
+
+
+def paginated_response(
+        crud_data: ListResponse[SchemaType], 
+        page: int, 
+        items_per_page: int
+) -> PaginatedListResponse[SchemaType]:
+    """
+    Create a paginated response based on the provided data and pagination parameters.
+
+    Parameters
+    ----------
+    crud_data : ListResponse[SchemaType]
+        Data to be paginated, including the list of items and total count.
+    page : int
+        Current page number.
+    items_per_page : int
+        Number of items per page.
+
+    Returns
+    -------
+    PaginatedListResponse[SchemaType]
+        A structured paginated response containing the list of items, total count, pagination flags, and numbers.
+
+    Note
+    ----
+    The function does not actually paginate the data but formats the response to indicate pagination metadata.
+    """
+    return {
+        "data": crud_data["data"],
+        "total_count": crud_data["total_count"],
+        "has_more": (page * items_per_page) < crud_data["total_count"],
+        "page": page,
+        "items_per_page": items_per_page
+    }
+
+def compute_offset(page: int, items_per_page: int) -> int:
+    """
+    Calculate the offset for pagination based on the given page number and items per page.
+
+    The offset represents the starting point in a dataset for the items on a given page.
+    For example, if each page displays 10 items and you want to display page 3, the offset will be 20,
+    meaning the display should start with the 21st item.
+
+    Parameters
+    ----------
+    page : int
+        The current page number. Page numbers should start from 1.
+    items_per_page : int
+        The number of items to be displayed on each page.
+
+    Returns
+    -------
+    int
+        The calculated offset.
+
+    Examples
+    --------
+    >>> offset(1, 10)
+    0
+    >>> offset(3, 10)
+    20
+    """
+    return (page - 1) * items_per_page

src/app/api/v1/posts.py

@@ -12,7 +12,7 @@
 from app.crud.crud_users import crud_users
 from app.api.exceptions import privileges_exception
 from app.core.cache import cache
-from app.core.models import PaginatedListResponse
+from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
 
 router = fastapi.APIRouter(tags=["posts"])
 
@@ -53,22 +53,20 @@ async def read_posts(
 
     posts_data = await crud_posts.get_multi(
         db=db,
-        offset=(page - 1) * items_per_page,
+        offset=compute_offset(page, items_per_page),
         limit=items_per_page,
         schema_to_select=PostRead,
         created_by_user_id=db_user["id"],
         is_deleted=False
     )
 
-    return {
-        "data": posts_data["data"],
-        "total_count": posts_data["total_count"],
-        "has_more": (page * items_per_page) < posts_data["total_count"],
-        "page": page,
-        "items_per_page": items_per_page
-    }
-
-
+    return paginated_response(
+        crud_data=posts_data, 
+        page=page, 
+        items_per_page=items_per_page
+    )
+    
+    
 @router.get("/{username}/post/{id}", response_model=PostRead)
 @cache(key_prefix="{username}_post_cache", resource_id_name="id")
 async def read_post(

src/app/api/v1/users.py

@@ -11,7 +11,7 @@
 from app.core.security import get_password_hash
 from app.crud.crud_users import crud_users
 from app.api.exceptions import privileges_exception
-from app.core.models import PaginatedListResponse
+from app.api.paginated import PaginatedListResponse, paginated_response, compute_offset
 
 router = fastapi.APIRouter(tags=["users"])
 
@@ -46,19 +46,17 @@ async def read_users(
 ):
     users_data = await crud_users.get_multi(
         db=db,
-        offset=(page - 1) * items_per_page,
+        offset=compute_offset(page, items_per_page),
         limit=items_per_page,
         schema_to_select=UserRead, 
         is_deleted=False
     )
 
-    return {
-        "data": users_data["data"],
-        "total_count": users_data["total_count"],
-        "has_more": (page * items_per_page) < users_data["total_count"],
-        "page": page,
-        "items_per_page": items_per_page
-    }
+    return paginated_response(
+        crud_data=users_data, 
+        page=page, 
+        items_per_page=items_per_page
+    )
 
 
 @router.get("/user/me/", response_model=UserRead)

src/app/core/models.py

@@ -1,23 +1,9 @@
-from typing import TypeVar, Generic, List
 import uuid as uuid_pkg
 from datetime import datetime
 
 from pydantic import BaseModel, Field, field_serializer
 from sqlalchemy import text
 
-ReadSchemaType = TypeVar("ReadSchemaType", bound=BaseModel)
-
-class ListResponse(BaseModel, Generic[ReadSchemaType]):
-    data: List[ReadSchemaType]
-
-
-class PaginatedListResponse(ListResponse[ReadSchemaType]):
-    total_count: int
-    has_more: bool
-    page: int | None = None
-    items_per_page: int | None = None
-
-
 class HealthCheck(BaseModel):
     name: str
     version: str