bdlg2023-sms-back/app/api/utils.py

import logging
from datetime import UTC, datetime
from uuid import UUID

from fastapi import HTTPException
from sqlalchemy import exists, select
from sqlalchemy.orm import Session
from sqlalchemy.sql import func

from app.core.email import EmailDeliveryError, EmailSender
from app.core.security import RESET_TOKEN_TTL, generate_reset_token
from app.models import PasswordResetToken, Project, User

logger = logging.getLogger(__name__)


def assert_project_exists_or_404(session: Session, project_id: UUID) -> None:
    if not session.execute(select(exists().where(Project.id == project_id))).scalar():
        raise HTTPException(status_code=404, detail="Project not found")


def get_project_or_404(session: Session, project_id: UUID) -> Project:
    p = session.get(Project, project_id)
    if p is None:
        raise HTTPException(status_code=404, detail="Project not found")
    return p


def get_project_organization_id(session: Session, project_id: UUID) -> str:
    """Fetches only Project.organization_id -- avoids loading the full
    Project row (and its relationships) just to check existence/ownership."""
    org_id = session.execute(
        select(Project.organization_id).where(Project.id == project_id)
    ).scalar_one_or_none()
    if org_id is None:
        raise HTTPException(status_code=404, detail="Project not found")
    return org_id


async def verify_id_list(
    session: Session,
    id_list: list[UUID],
    project_id: UUID,
    ObjectClass,
    error_message: str = "Invalid id list",
) -> None:
    """Verfiy the list of uuids exists and are from the right project

    ---
    Parameters
    - session : sqlachlemy Async session to use to verify slot validity
    - id_list : list of id to check
    - project_id :
    - ObjectClass : the ORM class of the object to check the ids from. Need to have id and project_id property

    ---
    Raise
    HTTPException(400, error_message) - if all the ids doesn't exists or are not associated with the right project
    """
    statement = select(func.count(ObjectClass.id)).where(
        ObjectClass.id.in_(id_list) & (ObjectClass.project_id == project_id)
    )
    results = session.execute(statement)
    if results.scalar() != len(id_list):
        raise HTTPException(status_code=400, detail=error_message)


def update_object_from_payload(obj, payload: dict):
    """Update the ORM model object from a pydantic payload dictionary"""
    for attr_name, value in payload.items():
        setattr(obj, attr_name, value)


async def issue_reset_token(
    session: Session, user: User, purpose: str, email_sender: EmailSender, base_url: str
) -> None:
    raw_token, token_hash = generate_reset_token()
    session.add(
        PasswordResetToken(
            user_id=user.id,
            token_hash=token_hash,
            expires_at=datetime.now(UTC) + RESET_TOKEN_TTL[purpose],
            purpose=purpose,
        )
    )
    session.commit()

    link = f"{base_url}/set-password?token={raw_token}"
    if purpose == "invite":
        subject, body = (
            "Vous avez été invité·e",
            f"Créez votre mot de passe : <a href='{link}'>{link}</a>",
        )
    else:
        subject, body = (
            "Réinitialisation du mot de passe",
            f"Réinitialisez ici : <a href='{link}'>{link}</a>",
        )
    try:
        await email_sender.send(user.email, subject, body)
    except EmailDeliveryError as exc:
        logger.error(
            "Failed to send %s email to user_id=%s email=%s: %s",
            purpose,
            user.id,
            user.email,
            exc,
        )