import uuid from typing import Annotated, Literal from fastapi import APIRouter, Depends, Request, Response from pydantic import BaseModel, ConfigDict from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.deps import get_current_user from app.auth.passwords import burn_verification_time, verify_password from app.auth.sessions import ( COOKIE_NAME, clear_session_cookie, create_auth_session, set_session_cookie, ) from app.db import get_db from app.errors import ApiError from app.models import AuthSession, User, UserRole router = APIRouter(prefix="/auth", tags=["auth"]) class LoginRequest(BaseModel): email: str password: str class UserOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID email: str name: str role: UserRole department_id: uuid.UUID | None # Pinned interface language, or null to follow the browser. Flows to the # UI via /me, so no page needs its own preference fetch. Typed as the # closed set the API accepts, so the generated client is precise too. locale: Literal["de", "en"] | None = None @router.post("/login") async def login( body: LoginRequest, response: Response, db: Annotated[AsyncSession, Depends(get_db)], ) -> UserOut: email = body.email.strip().lower() user = ( await db.execute(select(User).where(User.email == email)) ).scalar_one_or_none() if user is None: burn_verification_time() raise ApiError(401, "Invalid email or password.", "invalid_credentials") if not verify_password(user.password_hash, body.password): raise ApiError(401, "Invalid email or password.", "invalid_credentials") session = await create_auth_session(db, user) await db.commit() set_session_cookie(response, session) return UserOut.model_validate(user) @router.post("/logout", status_code=204) async def logout( request: Request, response: Response, db: Annotated[AsyncSession, Depends(get_db)], ) -> None: raw = request.cookies.get(COOKIE_NAME) if raw is not None: try: session_id = uuid.UUID(raw) except ValueError: session_id = None if session_id is not None: session = await db.get(AuthSession, session_id) if session is not None: await db.delete(session) await db.commit() clear_session_cookie(response) @router.get("/me") async def me(user: Annotated[User, Depends(get_current_user)]) -> UserOut: return UserOut.model_validate(user)