feat: org status check in auth dependencies

There is a hardcoded list of methods/endpoints for which the status check isn't done. i.e. the endpoints which need to be accessed before the org is approved.

Resolves #11
This commit is contained in:
Chris Milne 2026-05-28 10:55:39 +01:00
parent 4bf5933376
commit 9efd86cd5f
2 changed files with 25 additions and 10 deletions

View file

@ -9,35 +9,42 @@ Functions:
- List: Description - List: Description
- Functions: Description - Functions: Description
""" """
from typing import Annotated from typing import Annotated, Optional
from fastapi import Depends, Query from fastapi import Depends, Query, Request
from src.database import db_dependency from src.database import db_dependency
from src.organisation.schemas import OrgIDMixin from src.organisation.schemas import OrgIDMixin
from src.organisation.models import Organisation as Org from src.organisation.models import Organisation as Org
from src.organisation.exceptions import OrgNotFoundException from src.organisation.exceptions import OrgNotFoundException, AwaitingApprovalException
from src.organisation.constants import Status as OrgStatus
def get_org_model_query(db: db_dependency, org_id: Annotated[int, Query(gt=0)]) -> type[Org]: def get_org_model(db, request: Request, org_id: int):
org_model = db.get(Org, org_id) org_model = db.get(Org, org_id)
if org_model is None: if org_model is None:
raise OrgNotFoundException(org_id) raise OrgNotFoundException(org_id)
pre_approval_endpoints = ["PATCH/org/status", "PATCH/org/questionnaire", "GET/org/id"]
current_request = f"{request.method}{request.url.path}"
if current_request not in pre_approval_endpoints and org_model.status != OrgStatus.APPROVED:
raise AwaitingApprovalException(org_id)
return org_model return org_model
def get_org_model_query(db: db_dependency, request: Request, org_id: Annotated[int, Query(gt=0)]) -> type[Org]:
return get_org_model(db, request, org_id)
org_model_query_dependency = Annotated[type[Org], Depends(get_org_model_query)] org_model_query_dependency = Annotated[type[Org], Depends(get_org_model_query)]
def get_org_model_body(db: db_dependency, request_model: OrgIDMixin) -> type[Org]: def get_org_model_body(db: db_dependency, request: Request, request_model: OrgIDMixin) -> type[Org]:
org_id = getattr(request_model, "organisation_id", None) org_id: Optional[int] = getattr(request_model, "organisation_id", None)
if org_id is None: if org_id is None:
raise OrgNotFoundException raise OrgNotFoundException
org_model = db.get(Org, org_id)
if org_model is None:
raise OrgNotFoundException(org_id)
return org_model return get_org_model(db, request, org_id)
org_model_body_dependency = Annotated[type[Org], Depends(get_org_model_body)] org_model_body_dependency = Annotated[type[Org], Depends(get_org_model_body)]

View file

@ -17,3 +17,11 @@ class OrgNotFoundException(HTTPException):
status_code=status.HTTP_404_NOT_FOUND, status_code=status.HTTP_404_NOT_FOUND,
detail=detail, detail=detail,
) )
class AwaitingApprovalException(HTTPException):
def __init__(self, org_id: Optional[int] = None) -> None:
detail = "Organisation has not been approved." if org_id is None else f"Organisation with ID '{org_id}' has not been approved."
super().__init__(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
)