cloud-api/src/user/router.py

"""
Router endpoints for user module

Endpoints:
 - [get]/self/claims - Retrieves user's OIDC claims
 - [get]/self/db - Retrieves the user data from the db that corresponds to the current OIDC user
 - [get]/self/orgs - Retrieves all organisations associated with the current user
 - [get]/self/orgs/admin - Retrieves only admin organisations for the current user
 - [get]/{user_id} - Retrieves a specific user by their ID
 - [get]/{user_id}/orgs - Retrieves all organisations associated with a specific user
 - [get]/{user_id}/orgs/admin - Retrieves only admin organisations for a specific user
 - [delete]/{user_id} - Deletes a user from the db by their db ID
"""
from typing import Annotated

from fastapi import APIRouter
from fastapi.params import Path
from sqlalchemy.sql import exists
from starlette import status

from src.user.models import User
from src.user.schemas import UserResponse, OrgResponse, OIDCClaims
from src.user.exceptions import UserNotFoundException

from src.organisation.models import OrgUsers, Organisation

from src.auth.service import claims_dependency
from src.database import db_dependency

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


@router.get("/self/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", 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: claims_dependency, db: db_dependency):
	"""
	Returns the database details associated with the currently logged-in user.
	"""
	user_id = user.get("db_id", None)
	if user_id is None:
		raise UserNotFoundException()

	user_model = (db.query(User).filter(User.id == user_id).first())
	if user_model is None:
		raise UserNotFoundException(user_id=user_id)

	return user_model


@router.get("/self/orgs", response_model=list[OrgResponse], 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_current_organisations(db: db_dependency, user: claims_dependency):
	"""
	Returns all organisations associated with the currently logged-in user.
	"""
	user_id = user.get("db_id", None)
	if user_id is None:
		raise UserNotFoundException()
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise UserNotFoundException(user_id=user_id)

	org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
	                   .join(OrgUsers, Organisation.id == OrgUsers.org_id)
	                   .filter(OrgUsers.user_id == user_id)
	                   .all()
	                   )

	return org_user_models


@router.get("/self/orgs/admin", response_model=list[OrgResponse], 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_current_admin_organisations(db: db_dependency, user: claims_dependency):
	"""
	Returns the organisations for which the currently logged-in user is an admin.
	"""
	user_id = user.get("db_id", None)
	if user_id is None:
		raise UserNotFoundException()
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise UserNotFoundException(user_id=user_id)

	org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
	                   .join(OrgUsers, Organisation.id == OrgUsers.org_id)
	                   .filter(OrgUsers.user_id == user_id)
	                   .filter(OrgUsers.is_admin == True)
	                   .all()
	                   )

	return org_user_models


@router.get("/{user_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(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
	"""
	Returns the database details associated with the provided user ID.
	"""
	user_model = (db.query(User).filter(User.id == user_id).first())
	if user_model is None:
		raise UserNotFoundException(user_id=user_id)

	return user_model


@router.get("/{user_id}/orgs", response_model=list[OrgResponse], 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_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
	"""
	Returns all organisations associated with the provided user ID.
	"""
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise UserNotFoundException(user_id=user_id)

	org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
	                   .join(OrgUsers, Organisation.id == OrgUsers.org_id)
	                   .filter(OrgUsers.user_id == user_id)
	                   .all()
	                   )

	return org_user_models


@router.get("/{user_id}/orgs/admin", response_model=list[OrgResponse], 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_admin_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
	"""
	Returns the organisations for which the user with the provided user ID is an admin.
	"""
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise UserNotFoundException(user_id=user_id)

	org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
	                   .join(OrgUsers, Organisation.id == OrgUsers.org_id)
	                   .filter(OrgUsers.user_id == user_id)
	                   .filter(OrgUsers.is_admin == True)
	                   .all()
	                   )

	return org_user_models


@router.delete("/{user_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(user_id: Annotated[int, Path(gt=0)], db: db_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.
	"""
	user_model = (db.query(User).filter(User.id == user_id).first())
	if user_model is None:
		raise UserNotFoundException(user_id=user_id)
	db.delete(user_model)
	db.commit()