cloud-api/src/user/router.py

"""
Router endpoints for the user module

Endpoints:
 - [GET](/user/self/claims): [OIDC claims]: Returns all OIDC claims associated with the currently logged-in user.
 - [GET](/user/self/db): [OIDC claims]: Returns details about the currently logged-in user from the hub db.
 - [GET](/user/): [super admin]: Returns user(id) details.
 - [DELETE](/user/): [super admin]: Removes a User(id) from the hub database.
"""

from fastapi import APIRouter, status, BackgroundTasks

from src.iam.service import assign_default_user_group
from src.organisation.exceptions import OrgNotFoundException
from src.user.schemas import (
	UserResponse,
	OIDCClaims,
	UserPostInvitationRequest,
	UserPostInvitationAcceptRequest,
	UserGetSelfOrgsResponse,
	UserPostInvitationResponse,
	UserPostInvitationAcceptResponse,
)
from src.user.dependencies import (
	user_model_claims_dependency,
	user_model_query_dependency,
)
from src.user.service import send_invitation
from src.organisation.models import Organisation as Org

from src.auth.dependencies import (
	super_admin_dependency,
	org_model_root_claim_body_dependency,
)
from src.auth.service import claims_dependency
from src.database import db_dependency
from src.utils import verify_email_token

router = APIRouter(
	prefix="/user",
	tags=["User"],
)


@router.get(
	"/self/claims",
	summary="Get current user OIDC claims.",
	response_model=OIDCClaims,
	status_code=status.HTTP_200_OK,
	responses={
		status.HTTP_200_OK: {"description": "Successful retrieval from database"},
	},
)
async def current_user_claims(user: claims_dependency):
	"""
	Returns the full OIDC claims associated with the currently logged-in user.
	"""
	user["allowed_origins"] = user.get("allowed-origins", [])
	return user


@router.get(
	"/self/db",
	summary="Get current user hub details.",
	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 current_user(user_model: user_model_claims_dependency):
	"""
	Returns the database details associated with the currently logged-in user.
	"""
	return user_model


@router.get(
	"",
	summary="Get user hub details by ID.",
	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(
	user_model: user_model_query_dependency, su: super_admin_dependency
):
	"""
	Returns the database details associated with the provided user ID.
	"""
	return user_model


@router.delete(
	"",
	summary="Delete user from hub by ID.",
	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(
	db: db_dependency,
	user_model: user_model_query_dependency,
	su: super_admin_dependency,
):
	"""
	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.
	"""
	db.delete(user_model)
	db.commit()


@router.get(
	"/self/orgs",
	summary="Get all orgs the current user is a member of",
	status_code=status.HTTP_200_OK,
	response_model=UserGetSelfOrgsResponse,
	responses={},
)
async def get_user_orgs(user_model: user_model_claims_dependency):
	user_orgs = user_model.organisation_rel
	response = []
	for org in user_orgs:
		response.append(
			{
				"organisation_id": org.id,
				"name": org.name,
				"status": org.status,
				"intake_questionnaire": org.intake_questionnaire,
				"root_user_email": org.root_user_email,
				"billing_contact": {
					"id": org.billing_contact_id,
					"email": org.billing_contact_rel.email,
				},
				"owner_contact": {
					"id": org.owner_contact_id,
					"email": org.owner_contact_rel.email,
				},
				"security_contact": {
					"id": org.security_contact_id,
					"email": org.security_contact_rel.email,
				},
			}
		)

	return {"organisations": response}


@router.post(
	"/invitation",
	summary="Send an email invitation for a user to join an org",
	status_code=status.HTTP_200_OK,
	response_model=UserPostInvitationResponse,
)
async def invitation(
	background_tasks: BackgroundTasks,
	org_model: org_model_root_claim_body_dependency,
	request_model: UserPostInvitationRequest,
):
	org_id = org_model.id
	org_name = org_model.name
	user_email = request_model.user_email

	background_tasks.add_task(
		send_invitation, org_id=org_id, org_name=org_name, user_email=user_email
	)

	response = {
		"organisation": org_model,
		"invited_email": user_email,
	}

	return response


@router.post(
	"/invitation/accept",
	summary="Accept email invitation to join an org",
	status_code=status.HTTP_200_OK,
	response_model=UserPostInvitationAcceptResponse,
)
async def accept_invitation(
	db: db_dependency,
	user_model: user_model_claims_dependency,
	request_model: UserPostInvitationAcceptRequest,
):
	email_claims = await verify_email_token(
		token=request_model.jwt, user_model=user_model
	)

	org_model = db.get(Org, email_claims["org_id"])
	if org_model is None:
		raise OrgNotFoundException()

	org_model.user_rel.append(user_model)
	db.flush()
	await assign_default_user_group(db=db, org_model=org_model, user_model=user_model)

	response = {
		"organisation": org_model,
		"user": user_model,
	}

	db.commit()

	return response