cloud-api/src/user/service.py

"""
Module specific business logic for user module

Exports:
 - add_user_to_db: Creates a User record from OIDC claims, or updates user details
"""

from typing import Any
from datetime import datetime, timedelta, timezone

from sqlalchemy.orm import Session

from src.exceptions import UnprocessableContentException
from src.utils import send_email, generate_jwt

from src.user.schemas import OIDCUser
from src.user.models import User


async def add_user_to_db(db: Session, user_claims: dict[str, Any]) -> int:
	try:
		valid_user = OIDCUser(
			first_name=user_claims["given_name"],
			last_name=user_claims["family_name"],
			email=user_claims["email"],
			oidc_id=user_claims["sub"],
		)
	except Exception as e:
		print(e)
		raise UnprocessableContentException("Invalid or missing OIDC data")

	db_user = db.query(User).filter(User.oidc_id == valid_user.oidc_id).first()

	if not db_user:
		user_model = User(**valid_user.model_dump())
		db.add(user_model)
		user_id = user_model.id
		db.commit()
		return user_id
	else:
		user_id = db_user.id
		change = False
		if db_user.first_name != valid_user.first_name:
			db_user.first_name = valid_user.first_name
			change = True
		if db_user.last_name != valid_user.last_name:
			db_user.last_name = valid_user.last_name
			change = True
		if change:
			db.add(db_user)
			db.commit()
		return user_id


async def send_invitation(user_email: str, org_name: str, org_id: int):
	expiry_delta = timedelta(hours=24)
	expiry = datetime.now(timezone.utc) + expiry_delta
	claims = {
		"email": user_email,
		"org_id": org_id,
		"exp": expiry,
		"type": "org_invite",
	}

	token = await generate_jwt(claims)
	subject = f"You have been invited to join {org_name}"
	body = f"You have been invited to join {org_name}.\nClick the link to accept.\nfrontend.capture/send/to/endpoint/{token}"

	await send_email(
		recipient=user_email,
		subject=subject,
		body=body,
	)