63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
"""
|
|
Module specific business logic for user module
|
|
"""
|
|
|
|
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(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:
|
|
logging.exception(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
|
|
|
|
user_id = db_user.id
|
|
db_user.first_name = valid_user.first_name
|
|
db_user.last_name = valid_user.last_name
|
|
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,
|
|
)
|