feat: org status check moved
Accessing endpoints as super admin no longer requires the org to be approved.
This commit is contained in:
parent
a655eaf543
commit
092e12a892
4 changed files with 94 additions and 115 deletions
|
|
@ -7,65 +7,38 @@ Exports:
|
|||
"""
|
||||
|
||||
from typing import Annotated, Optional
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from fastapi import Depends, Query, Request
|
||||
from fastapi import Depends, Query
|
||||
|
||||
from src.database import db_dependency
|
||||
from src.exceptions import ForbiddenException
|
||||
|
||||
from src.organisation.schemas import OrgIDMixin
|
||||
from src.organisation.models import Organisation as Org
|
||||
from src.organisation.exceptions import OrgNotFoundException, AwaitingApprovalException
|
||||
from src.organisation.constants import Status as OrgStatus
|
||||
|
||||
|
||||
def get_org_model(db: Session, request: Request, org_id: int):
|
||||
org_model = db.get(Org, org_id)
|
||||
if org_model is None:
|
||||
raise OrgNotFoundException(org_id)
|
||||
|
||||
org_status = OrgStatus(org_model.status)
|
||||
if org_status.is_blocked:
|
||||
raise ForbiddenException("This organisation cannot perform this action.")
|
||||
|
||||
root = "/api/v1"
|
||||
|
||||
pre_approval_endpoints = [
|
||||
f"PATCH{root}/org/status",
|
||||
f"PATCH{root}/org/questionnaire",
|
||||
f"GET{root}/org",
|
||||
f"GET{root}/org/contact",
|
||||
f"PATCH{root}/org/contact",
|
||||
f"DELETE{root}/org/self",
|
||||
]
|
||||
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
|
||||
from src.organisation.exceptions import OrgNotFoundException
|
||||
|
||||
|
||||
def get_org_model_query(
|
||||
db: db_dependency, request: Request, org_id: Annotated[int, Query(gt=0)]
|
||||
db: db_dependency, org_id: Annotated[int, Query(gt=0)]
|
||||
) -> type[Org]:
|
||||
return get_org_model(db, request, org_id)
|
||||
org_model = db.get(Org, org_id)
|
||||
if org_model is None:
|
||||
raise OrgNotFoundException(org_id)
|
||||
return org_model
|
||||
|
||||
|
||||
org_model_query_dependency = Annotated[type[Org], Depends(get_org_model_query)]
|
||||
|
||||
|
||||
def get_org_model_body(
|
||||
db: db_dependency, request: Request, request_model: OrgIDMixin
|
||||
) -> type[Org]:
|
||||
def get_org_model_body(db: db_dependency, request_model: OrgIDMixin) -> type[Org]:
|
||||
org_id: Optional[int] = getattr(request_model, "organisation_id", None)
|
||||
if org_id is None:
|
||||
raise OrgNotFoundException()
|
||||
|
||||
return get_org_model(db, request, org_id)
|
||||
org_model = db.get(Org, org_id)
|
||||
if org_model is None:
|
||||
raise OrgNotFoundException(org_id)
|
||||
|
||||
return org_model
|
||||
|
||||
|
||||
org_model_body_dependency = Annotated[type[Org], Depends(get_org_model_body)]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue