1
0
Fork 0
forked from sr2/cloud-api

feat: user dependencies

In addition to the by-query and by-body db fetch dependencies. Users also have a by-claim dependency.

Issue #6
This commit is contained in:
Chris Milne 2026-05-27 13:27:24 +01:00
parent 657f91d73d
commit 748544fb82
3 changed files with 59 additions and 27 deletions

View file

@ -8,4 +8,47 @@ Classes:
Functions:
- List: Description
- Functions: Description
"""
"""
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)]

View file

@ -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()

View file

@ -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