Paginate users list endpoint

README.md

@@ -116,6 +116,8 @@ following advantages to starting your own from scratch :
   very easily, add users (and make admin), and run a development server. This
   can easily be modified to add your own functionality (for example bulk add
   data) since it is based on the excellent [Typer][typer] library.
+- Built-in pagination support for user listing and search endpoints, using
+  `fastapi-pagination`.
 - Easily batch-add random test users to the database for testing/development
   purposes using the CLI or seed the database with pre-set users from a CSV
   file.

TODO.md

@@ -25,9 +25,6 @@
   this should be implemented. *This may just need to be in derived projects
   though, not this template*.
 - Add Metrics and Observability (eg Prometheus, Grafana, Sentry, etc)
-- Add pagination to the user list endpoint. Implement this in a way that is
-  generic and can be used for other custom endpoints too. The library
-  'fastapi-pagination' is really good and performant.
 - Use an alternative logger if uvicorn is not being used for some reason.
 - Allow the option of using auto-incerement integers for the User ID (as it is
   now) or UUID's which is more secure and common these days. This should be

app/managers/user.py

@@ -366,6 +366,11 @@ async def get_all_users(session: AsyncSession) -> Sequence[User]:
         """Get all Users."""
         return await get_all_users_(session)
 
+    @staticmethod
+    def list_users_query() -> Select[tuple[User]]:
+        """Return the base users query for pagination."""
+        return select(User).order_by(User.id)
+
     @staticmethod
     async def get_user_by_id(user_id: int, session: AsyncSession) -> User:
         """Return one user by ID."""

app/resources/user.py

@@ -1,10 +1,9 @@
 """Routes for User listing and control."""
 
-from collections.abc import Sequence
 from typing import Annotated, cast
 
 from fastapi import APIRouter, Depends, Request, status
-from fastapi_pagination import Page
+from fastapi_pagination import Page, Params
 from fastapi_pagination.ext.sqlalchemy import apaginate
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -14,6 +13,7 @@
 from app.managers.user import UserManager
 from app.models.enums import RoleType
 from app.models.user import User
+from app.schemas.examples import ExampleUser
 from app.schemas.request.user import (
     SearchField,
     UserChangePasswordRequest,
@@ -23,25 +23,63 @@
 
 router = APIRouter(tags=["Users"], prefix="/users")
 
+USER_EXAMPLE = {
+    "id": ExampleUser.id,
+    "first_name": ExampleUser.first_name,
+    "last_name": ExampleUser.last_name,
+    "email": ExampleUser.email,
+    "role": ExampleUser.role,
+    "banned": ExampleUser.banned,
+    "verified": ExampleUser.verified,
+}
+
+PAGINATED_USERS_EXAMPLE = {
+    "items": [USER_EXAMPLE],
+    "total": 1,
+    "page": 1,
+    "size": 50,
+    "pages": 1,
+}
+
 
 @router.get(
     "/",
     dependencies=[Depends(get_current_user), Depends(is_admin)],
-    response_model=UserResponse | list[UserResponse],
+    response_model=UserResponse | Page[UserResponse],
+    responses={
+        200: {
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "single_user": {
+                            "summary": "Single user",
+                            "value": USER_EXAMPLE,
+                        },
+                        "paginated_users": {
+                            "summary": "Paginated users",
+                            "value": PAGINATED_USERS_EXAMPLE,
+                        },
+                    }
+                }
+            }
+        }
+    },
 )
 async def get_users(
     db: Annotated[AsyncSession, Depends(get_database)],
+    params: Annotated[Params, Depends()],
     user_id: int | None = None,
-) -> Sequence[User] | User:
-    """Get all users or a specific user by their ID.
+) -> Page[UserResponse] | User:
+    """Get all users (paginated) or a specific user by their ID.
 
     user_id is optional, and if omitted then all Users are returned.
 
     This route is only allowed for Admins.
     """
     if user_id:
         return await UserManager.get_user_by_id(user_id, db)
-    return await UserManager.get_all_users(db)
+    query = UserManager.list_users_query()
+    return cast("Page[UserResponse]", await apaginate(db, query, params))
 
 
 @router.get(

tests/integration/test_user_routes.py

@@ -100,7 +100,9 @@ async def test_admin_can_get_all_users(
         )
 
         assert response.status_code == status.HTTP_200_OK
-        assert len(response.json()) == 4  # noqa: PLR2004
+        payload = response.json()
+        assert payload["total"] == 4  # noqa: PLR2004
+        assert len(payload["items"]) == 4  # noqa: PLR2004
 
     async def test_admin_can_get_one_user(
         self, client: AsyncClient, test_db: AsyncSession