feat: custom exceptions instead of direct fastapi.httpexceptions

Resolves #2
This commit is contained in:
Chris Milne 2026-05-27 14:58:10 +01:00
parent d3d3b2ca63
commit 868e56ce40
9 changed files with 73 additions and 43 deletions

View file

@ -10,18 +10,19 @@ Functions:
- Functions: Description
"""
from typing import Annotated, Any
from fastapi import Depends, HTTPException
from fastapi import Depends
from src.user.dependencies import user_model_claims_dependency
from src.organisation.dependencies import org_model_query_dependency
from src.auth.exceptions import UnauthorizedException
async def org_query_user_claims(org_model: org_model_query_dependency, user_model: user_model_claims_dependency):
if user_model in org_model.user_rel:
return True
raise HTTPException(status_code=401, detail="Not authorised")
raise UnauthorizedException()
org_query_user_claims_dependency = Annotated[dict[str, Any], Depends(org_query_user_claims)]
@ -31,7 +32,7 @@ async def org_query_root_claims(user_model: user_model_claims_dependency, org_mo
if org_model.root_user_id == user_model.id:
return True
raise HTTPException(status_code=401, detail="Not authorised")
raise UnauthorizedException()
org_query_root_claims_dependency = Annotated[dict[str, Any], Depends(org_query_root_claims)]
@ -40,8 +41,7 @@ org_query_root_claims_dependency = Annotated[dict[str, Any], Depends(org_query_r
async def is_super_admin(user_model: user_model_claims_dependency):
super_admin_emails = []
if user_model.email not in super_admin_emails:
raise HTTPException(status_code=401, detail="Not authorised")
raise UnauthorizedException()
return True

View file

@ -4,4 +4,16 @@ Module specific exceptions for auth module
Exceptions:
- List: Description
- Exceptions: Description
"""
"""
from typing import Optional
from fastapi import HTTPException, status
class UnauthorizedException(HTTPException):
def __init__(self, message: Optional[str] = None) -> None:
detail = "Not authorized" if not message else message
super().__init__(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
)

View file

@ -13,9 +13,10 @@ from joserfc.errors import ExpiredTokenError
from joserfc.jwk import KeySet
from urllib.request import urlopen
from fastapi import Depends, HTTPException
from fastapi import Depends
from fastapi.security import OpenIdConnect
from src.auth.exceptions import UnauthorizedException
from src.auth.config import auth_settings
from src.user.service import add_user_to_db
@ -50,8 +51,7 @@ async def get_current_user(oidc_auth_string: oidc_dependency) -> dict[str, Any]:
try:
claims_requests.validate(token.claims)
except ExpiredTokenError:
raise HTTPException(status_code=401, detail="Token expired")
raise UnauthorizedException(message="Token is expired")
db_id = await add_user_to_db(token.claims)
token.claims["db_id"] = db_id