cloud-api/src/organisation/dependencies.py

"""
Dependencies related to the organisation module

Exports:
 - org_model_query_dependency: org_model: Gets org model from db, if it exists. Uses org_id from query param. Also verifies if the org has been approved.
 - org_model_body_dependency: org_model: Gets org model from db, if it exists. Uses org_id from request body. Also verifies if the org has been approved.
"""

from typing import Annotated, Optional
from sqlalchemy.orm import Session

from fastapi import Depends, Query, Request

from src.database import db_dependency

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)

	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


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)]


def get_org_model_body(
	db: db_dependency, request: Request, 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_body_dependency = Annotated[type[Org], Depends(get_org_model_body)]