Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions app/managers/user.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@staticmethod
async def get_user_by_id(user_id: int, session: AsyncSession) -> User:
"""Return one user by ID."""
Expand Down
50 changes: 44 additions & 6 deletions app/resources/user.py
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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,
Expand All @@ -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(
Expand Down
4 changes: 3 additions & 1 deletion tests/integration/test_user_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down