luxferre
c689ac1e10
All changes are either: - Correcting tabs - Adding/removing line breaks - Adding trailing commas
50 lines
1.2 KiB
Python
50 lines
1.2 KiB
Python
"""
|
|
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 sqlalchemy.orm import Session
|
|
|
|
from src.exceptions import UnprocessableContentException
|
|
|
|
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
|