diff --git a/src/user/dependencies.py b/src/user/dependencies.py index 525faf2..c5314d0 100644 --- a/src/user/dependencies.py +++ b/src/user/dependencies.py @@ -8,4 +8,47 @@ Classes: Functions: - List: Description - Functions: Description -""" \ No newline at end of file +""" +from typing import Annotated +from fastapi import Depends, Query + +from src.user.exceptions import UserNotFoundException +from src.user.models import User + +from src.auth.service import claims_dependency +from src.database import db_dependency +from src.user.schemas import UserIDMixin + + +async def get_user_model_claims(claims: claims_dependency, db: db_dependency): + user_id = claims.get("db_id", None) + if user_id is None: + raise UserNotFoundException() + + user_model = db.get(User, user_id) + if user_model is None: + raise UserNotFoundException(user_id=user_id) + + return user_model + +user_model_claims_dependency = Annotated[type[User], Depends(get_user_model_claims)] + + +async def get_user_model_query(db: db_dependency, user_id: Annotated[int, Query(gt=0)]): + user_model = db.get(User, user_id) + if user_model is None: + raise UserNotFoundException(user_id=user_id) + + return user_model + +user_model_query_dependency = Annotated[type[User], Depends(get_user_model_query)] + + +async def get_user_model_body(db: db_dependency, request_model: UserIDMixin): + user_model = db.get(User, request_model.user_id) + if user_model is None: + raise UserNotFoundException(user_id=request_model.user_id) + + return user_model + +user_model_body_dependency = Annotated[type[User], Depends(get_user_model_body)] diff --git a/src/user/router.py b/src/user/router.py index 6d7a5a8..00d619b 100644 --- a/src/user/router.py +++ b/src/user/router.py @@ -11,15 +11,11 @@ Endpoints: - [get]/{user_id}/orgs/admin - Retrieves only admin organisations for a specific user - [delete]/{user_id} - Deletes a user from the db by their db ID """ -from typing import Annotated - from fastapi import APIRouter -from fastapi.params import Path from starlette import status -from src.user.models import User -from src.user.schemas import UserResponse, OIDCClaims -from src.user.exceptions import UserNotFoundException +from src.user.schemas import UserResponse, OIDCClaims, UserDeleteUserRequest +from src.user.dependencies import user_model_claims_dependency, user_model_query_dependency, user_model_body_dependency from src.auth.service import claims_dependency from src.database import db_dependency @@ -45,46 +41,31 @@ async def current_user_claims(user: claims_dependency): 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): +async def current_user(user_model: user_model_claims_dependency): """ 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 == user_id).first()) - if user_model is None: - raise UserNotFoundException(user_id=user_id) - return user_model -@router.get("/{user_id}", response_model=UserResponse, status_code=status.HTTP_200_OK, responses={ +@router.get("/", 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")]): +async def get_user_by_id(user_model: user_model_query_dependency): """ 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 UserNotFoundException(user_id=user_id) - return user_model -@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT, responses={ +@router.delete("/", 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): +async def delete_user_by_id(db: db_dependency, user_model: user_model_body_dependency, request_model: UserDeleteUserRequest): """ 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 UserNotFoundException(user_id=user_id) db.delete(user_model) db.commit() diff --git a/src/user/schemas.py b/src/user/schemas.py index 7e3dc12..285f510 100644 --- a/src/user/schemas.py +++ b/src/user/schemas.py @@ -9,6 +9,10 @@ from typing import Optional from src.schemas import CustomBaseModel +class UserIDMixin(CustomBaseModel): + user_id: int + + class OIDCClaims(CustomBaseModel): exp: int iat: int @@ -52,3 +56,7 @@ class UserResponse(CustomBaseModel): class OrgResponse(CustomBaseModel): org_id: int name: str + + +class UserDeleteUserRequest(UserIDMixin): + pass \ No newline at end of file