cloud-api/src/user/router.py

"""
Router endpoints for user module

Endpoints:
 - [get]/me/claims - Retrieves user's OIDC claims
 - [get]/me/db - Retrieves the user data from the db that corresponds to the current OIDC user
 - [get]/me/orgs - Retrieves all organisations associated with the current user
 - [get]/me/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 fastapi import APIRouter, HTTPException
from fastapi.params import Path
from sqlalchemy.sql import exists

from src.user.models import User
from src.user.schemas import UserResponse, OIDCUser, OrgResponse

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("/me/claims")
async def current_user_claims(user: claims_dependency):
	return user


@router.get("/me/db", response_model=OIDCUser)
async def current_user(user: claims_dependency, db: db_dependency):
	db_id = user.get("db_id", None)
	if db_id is None:
		raise HTTPException(status_code=404, detail="User not found in db")

	user_model = (db.query(User).filter(User.id == db_id).first())
	if user_model is None:
		raise HTTPException(status_code=404, detail="User not found")

	return user_model


@router.get("/me/orgs", response_model=list[OrgResponse])
async def get_current_organisations(db: db_dependency, user: claims_dependency):
	user_id = user.get("db_id", None)
	if user_id is None:
		raise HTTPException(status_code=404, detail="User not found")
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise HTTPException(status_code=404, detail="User not found")

	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("/me/orgs/admin", response_model=list[OrgResponse])
async def get_current_admin_organisations(db: db_dependency, user: claims_dependency):
	user_id = user.get("db_id", None)
	if user_id is None:
		raise HTTPException(status_code=404, detail="User not found")
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise HTTPException(status_code=404, detail="User not found")

	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)
async def get_user_by_id(user_id: int, db: db_dependency):
	user_model = (db.query(User).filter(User.id == user_id).first())
	if user_model is None:
		raise HTTPException(status_code=404, detail="User not found")

	return user_model


@router.get("/{user_id}/orgs", response_model=list[OrgResponse])
async def get_organisations(db: db_dependency, user_id: int = Path(gt=0)):
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise HTTPException(status_code=404, detail="User not found")

	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])
async def get_admin_organisations(db: db_dependency, user_id: int = Path(gt=0)):
	user_exists = db.query(exists().where(User.id == user_id)).scalar()
	if not user_exists:
		raise HTTPException(status_code=404, detail="User not found")

	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}")
async def delete_user_by_id(user_id: int, db: db_dependency):
	user_model = (db.query(User).filter(User.id == user_id).first())
	if user_model is None:
		raise HTTPException(status_code=404, detail="User not found")
	db.delete(user_model)
	db.commit()