From 83a24a91f4099a7f7879065bd501cbabf5e5913f Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 20 May 2026 15:23:40 +0100 Subject: [PATCH] docs: user module In-line and Swagger docs improvements on the User module and endpoints --- src/main.py | 11 +++- src/user/exceptions.py | 14 ++++- src/user/router.py | 114 +++++++++++++++++++++++++++++------------ 3 files changed, 105 insertions(+), 34 deletions(-) diff --git a/src/main.py b/src/main.py index 272e5a6..4b4d1a9 100644 --- a/src/main.py +++ b/src/main.py @@ -26,12 +26,21 @@ if settings.ENVIRONMENT.is_deployed: pass +tags_metadata = [ + { + "name": "User", + "description": "User related operations, includes getting information about the current user", + } +] + + app = FastAPI( swagger_ui_init_oauth={ "clientId": auth_settings.CLIENT_ID, "usePkceWithAuthorizationCodeGrant": True, "scopes": "openid profile email", - } + }, + openapi_tags=tags_metadata, ) # Type inspection disabled for middleware injection. diff --git a/src/user/exceptions.py b/src/user/exceptions.py index 5c1087e..6f2a669 100644 --- a/src/user/exceptions.py +++ b/src/user/exceptions.py @@ -4,4 +4,16 @@ Module specific exceptions for user module Exceptions: - List: Description - Exceptions: Description -""" \ No newline at end of file +""" +from typing import Optional + +from fastapi import HTTPException, status + + +class UserNotFoundException(HTTPException): + def __init__(self, user_id: Optional[int] = None) -> None: + detail = "User not found" if user_id is None else f"User with ID '{user_id}' was not found." + super().__init__( + status_code=status.HTTP_404_NOT_FOUND, + detail=detail, + ) diff --git a/src/user/router.py b/src/user/router.py index 40873ec..fb2b632 100644 --- a/src/user/router.py +++ b/src/user/router.py @@ -2,10 +2,10 @@ Router endpoints for user module Endpoints: - - [get]/me/claims - Retrieves user's OIDC claims - - [get]/me/db - Retrieves the user data from the db that corresponds to the current OIDC user - - [get]/me/orgs - Retrieves all organisations associated with the current user - - [get]/me/orgs/admin - Retrieves only admin organisations for the current user + - [get]/self/claims - Retrieves user's OIDC claims + - [get]/self/db - Retrieves the user data from the db that corresponds to the current OIDC user + - [get]/self/orgs - Retrieves all organisations associated with the current user + - [get]/self/orgs/admin - Retrieves only admin organisations for the current user - [get]/{user_id} - Retrieves a specific user by their ID - [get]/{user_id}/orgs - Retrieves all organisations associated with a specific user - [get]/{user_id}/orgs/admin - Retrieves only admin organisations for a specific user @@ -13,12 +13,14 @@ Endpoints: """ from typing import Annotated -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter from fastapi.params import Path from sqlalchemy.sql import exists +from starlette import status from src.user.models import User -from src.user.schemas import UserResponse, OIDCUser, OrgResponse +from src.user.schemas import UserResponse, OrgResponse, OIDCClaims +from src.user.exceptions import UserNotFoundException from src.organisation.models import OrgUsers, Organisation @@ -27,36 +29,54 @@ from src.database import db_dependency router = APIRouter( prefix="/user", - tags=["user"], + tags=["User"], ) -@router.get("/me/claims") +@router.get("/self/claims", response_model=OIDCClaims, status_code=status.HTTP_200_OK, responses={ + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) async def current_user_claims(user: claims_dependency): + """ + Returns the full OIDC claims associated with the currently logged-in user. + """ + user["allowed_origins"] = user.get("allowed-origins", []) return user -@router.get("/me/db", response_model=OIDCUser) +@router.get("/self/db", response_model=UserResponse, status_code=status.HTTP_200_OK, responses={ + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) async def current_user(user: claims_dependency, db: db_dependency): - db_id = user.get("db_id", None) - if db_id is None: - raise HTTPException(status_code=404, detail="User not found in db") + """ + Returns the database details associated with the currently logged-in user. + """ + user_id = user.get("db_id", None) + if user_id is None: + raise UserNotFoundException() - user_model = (db.query(User).filter(User.id == db_id).first()) + user_model = (db.query(User).filter(User.id == user_id).first()) if user_model is None: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) return user_model -@router.get("/me/orgs", response_model=list[OrgResponse]) +@router.get("/self/orgs", response_model=list[OrgResponse], status_code=status.HTTP_200_OK, responses={ + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) async def get_current_organisations(db: db_dependency, user: claims_dependency): + """ + Returns all organisations associated with the currently logged-in user. + """ user_id = user.get("db_id", None) if user_id is None: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException() user_exists = db.query(exists().where(User.id == user_id)).scalar() if not user_exists: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name) .join(OrgUsers, Organisation.id == OrgUsers.org_id) @@ -67,14 +87,20 @@ async def get_current_organisations(db: db_dependency, user: claims_dependency): return org_user_models -@router.get("/me/orgs/admin", response_model=list[OrgResponse]) +@router.get("/self/orgs/admin", response_model=list[OrgResponse], status_code=status.HTTP_200_OK, responses={ + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) async def get_current_admin_organisations(db: db_dependency, user: claims_dependency): + """ + Returns the organisations for which the currently logged-in user is an admin. + """ user_id = user.get("db_id", None) if user_id is None: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException() user_exists = db.query(exists().where(User.id == user_id)).scalar() if not user_exists: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name) .join(OrgUsers, Organisation.id == OrgUsers.org_id) @@ -86,20 +112,32 @@ async def get_current_admin_organisations(db: db_dependency, user: claims_depend return org_user_models -@router.get("/{user_id}", response_model=UserResponse) -async def get_user_by_id(user_id: int, db: db_dependency): +@router.get("/{user_id}", response_model=UserResponse, status_code=status.HTTP_200_OK, responses={ + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) +async def get_user_by_id(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]): + """ + Returns the database details associated with the provided user ID. + """ user_model = (db.query(User).filter(User.id == user_id).first()) if user_model is None: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) return user_model -@router.get("/{user_id}/orgs", response_model=list[OrgResponse]) -async def get_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0)]): +@router.get("/{user_id}/orgs", response_model=list[OrgResponse], status_code=status.HTTP_200_OK, responses={ + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) +async def get_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]): + """ + Returns all organisations associated with the provided user ID. + """ user_exists = db.query(exists().where(User.id == user_id)).scalar() if not user_exists: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name) .join(OrgUsers, Organisation.id == OrgUsers.org_id) @@ -110,11 +148,17 @@ async def get_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0 return org_user_models -@router.get("/{user_id}/orgs/admin", response_model=list[OrgResponse]) -async def get_admin_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0)]): +@router.get("/{user_id}/orgs/admin", response_model=list[OrgResponse], status_code=status.HTTP_200_OK, responses={ + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + status.HTTP_200_OK: {"description": "Successful retrieval from database"}, +}) +async def get_admin_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]): + """ + Returns the organisations for which the user with the provided user ID is an admin. + """ user_exists = db.query(exists().where(User.id == user_id)).scalar() if not user_exists: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name) .join(OrgUsers, Organisation.id == OrgUsers.org_id) @@ -126,10 +170,16 @@ async def get_admin_organisations(db: db_dependency, user_id: Annotated[int, Pat return org_user_models -@router.delete("/{user_id}") -async def delete_user_by_id(user_id: int, db: db_dependency): +@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT, responses={ + status.HTTP_204_NO_CONTENT: {"description": "User deleted"}, + status.HTTP_404_NOT_FOUND: {"description": "User not found"}, + }) +async def delete_user_by_id(user_id: Annotated[int, Path(gt=0)], db: db_dependency): + """ + Deletes the user with the provided ID from the database. This will not remove them from OIDC, and they will be automatically readded on next login. + """ user_model = (db.query(User).filter(User.id == user_id).first()) if user_model is None: - raise HTTPException(status_code=404, detail="User not found") + raise UserNotFoundException(user_id=user_id) db.delete(user_model) db.commit()