Paginate GET /registrations using fastapi-pagination

fideslog/api/database/registrations.py

@@ -1,7 +1,8 @@
 from datetime import datetime, timezone
 from logging import getLogger
-from typing import List, Optional
 
+from fastapi_pagination.bases import AbstractPage, AbstractParams
+from fastapi_pagination.ext.sqlalchemy import paginate
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import Session
 
@@ -13,18 +14,17 @@
 
 def get(
     database: Session,
-    count: Optional[int],
-    offset: Optional[int],
-) -> List[RegistrationORM]:
+    pagination_params: AbstractParams,
+) -> AbstractPage[RegistrationORM]:
     """
     Return existing registrations.
     """
 
     log.debug("Fetching registrations")
-    if count is not None:
-        return database.query(RegistrationORM).limit(count).offset(offset or 0).all()
-
-    return database.query(RegistrationORM).all()
+    return paginate(
+        database.query(RegistrationORM).order_by(RegistrationORM.created_at.desc()),
+        params=pagination_params,
+    )
 
 
 def create(database: Session, registration: Registration) -> None:

fideslog/api/requirements.txt

@@ -1,5 +1,6 @@
-fastapi==0.74.0
-pydantic[email]==1.9.0
+fastapi-pagination[sqlalchemy]== 0.10.0
+fastapi==0.82.0
+pydantic[email]==1.9.1
 slowapi==0.1.5
 snowflake-sqlalchemy==1.3.3
 SQLAlchemy-Utils==0.38.3

fideslog/api/routes/registrations.py

@@ -1,14 +1,17 @@
 from logging import getLogger
-from typing import List, Optional
+from typing import List
 
-from fastapi import APIRouter, Depends, Query, Request, Response, status
+from fastapi import APIRouter, Depends, Request, Response, status
+from fastapi_pagination import Params
+from fastapi_pagination.bases import AbstractPage
 from sqlalchemy.exc import DBAPIError, NoResultFound
 from sqlalchemy.orm import Session
 from sqlalchemy.orm.exc import UnmappedInstanceError
 
 from ..database import get_db
 from ..database.registrations import create, delete, get, update
 from ..errors import InternalServerError, NotFoundError, TooManyRequestsError
+from ..models.models import Registration as RegistrationORM
 from ..schemas.registration import Registration
 
 log = getLogger(__name__)
@@ -27,37 +30,19 @@
 )
 async def list_registrations(
     _: Request,
-    count: Optional[int] = Query(
-        default=None,
-        description="The amount of registrations to return.",
-        gt=0,
-    ),
-    offset: Optional[int] = Query(
-        default=None,
-        description="The amount of registrations to skip before returning.",
-        gt=0,
-    ),
+    params: Params = Depends(),
     database: Session = Depends(get_db),
-) -> List[Registration]:
+) -> AbstractPage[RegistrationORM]:
     """
     List existing registrations.
     """
 
     try:
-        registrations = get(database, count, offset)
+        registrations = get(database, params)
     except DBAPIError as err:
         raise InternalServerError(err) from err
 
-    return [
-        Registration(
-            client_id=registration.client_id,
-            email=registration.email,
-            organization=registration.organization,
-            created_at=registration.created_at,
-            updated_at=registration.updated_at,
-        )
-        for registration in registrations
-    ]
+    return registrations.items  # type: ignore[attr-defined]
 
 
 @registration_router.post(
@@ -102,7 +87,7 @@ async def modify_registration(
     _: Request,
     registration: Registration,
     database: Session = Depends(get_db),
-) -> Registration:
+) -> RegistrationORM:
     """
     Update an existing registration.
     """
@@ -114,13 +99,7 @@ async def modify_registration(
     except Exception as err:
         raise InternalServerError(err) from err
 
-    return Registration(
-        client_id=updated.client_id,
-        email=updated.email,
-        organization=updated.organization,
-        created_at=updated.created_at,
-        updated_at=updated.updated_at,
-    )
+    return updated
 
 
 @registration_router.delete(