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.auth.exceptions import UnauthorizedException
from src.organisation.exceptions import OrgNotFoundException
from src.user.schemas import (
	UserResponse,
	OIDCClaims,
	UserPostInvitationRequest,
	UserPostInvitationAcceptRequest,
)
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 decode_jwt

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.post(
	"/invitation",
	summary="Send an email invitation for a user to join an org",
	status_code=status.HTTP_200_OK,
)
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
	)

	return "Invitation sent"


@router.post(
	"/invitation/accept",
	summary="Accept email invitation to join an org",
	status_code=status.HTTP_200_OK,
)
async def accept_invitation(
	db: db_dependency,
	user_model: user_model_claims_dependency,
	request_model: UserPostInvitationAcceptRequest,
):
	email_claims = await decode_jwt(request_model.jwt)
	claimed_email = email_claims["user_email"]

	if user_model.email != claimed_email:
		raise UnauthorizedException("The logged in user and email do not match.")

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

	org_model.user_rel.append(user_model)
	db.commit()

	return "Invitation accepted"