"""User service for permanent user information storage This service handles upsert operations for the users table, which stores display names and metadata for report generation. """ from datetime import datetime from typing import Optional from sqlalchemy.orm import Session from app.modules.auth.models import User def upsert_user( db: Session, user_id: str, display_name: str, office_location: Optional[str] = None, job_title: Optional[str] = None, ) -> User: """Create or update user record with AD information This function is called on every successful login to keep user information up to date. Uses SQLAlchemy merge for atomic upsert operation. Args: db: Database session user_id: User email address (primary key) display_name: Display name from AD API office_location: Office location from AD API (optional) job_title: Job title from AD API (optional) Returns: User: The created or updated user record """ # Check if user exists existing_user = db.query(User).filter(User.user_id == user_id).first() if existing_user: # Update existing user existing_user.display_name = display_name existing_user.office_location = office_location existing_user.job_title = job_title existing_user.last_login_at = datetime.utcnow() db.commit() db.refresh(existing_user) return existing_user else: # Create new user new_user = User( user_id=user_id, display_name=display_name, office_location=office_location, job_title=job_title, last_login_at=datetime.utcnow(), created_at=datetime.utcnow(), ) db.add(new_user) db.commit() db.refresh(new_user) return new_user def get_user_by_id(db: Session, user_id: str) -> Optional[User]: """Get user by user_id (email) Args: db: Database session user_id: User email address Returns: User or None if not found """ return db.query(User).filter(User.user_id == user_id).first() def get_display_name(db: Session, user_id: str) -> str: """Get display name for a user, falling back to email if not found Args: db: Database session user_id: User email address Returns: Display name or email address as fallback """ user = get_user_by_id(db, user_id) if user: return user.display_name return user_id # Fallback to email if user not in database