import re from fastapi import APIRouter, Depends, HTTPException, Response from sqlalchemy.orm import Session from ..db import get_db from ..models import Role, User from ..security import hash_password, make_session_value, set_session router = APIRouter(prefix="/setup", tags=["setup"]) def _has_admin(db: Session) -> bool: return db.query(User).filter(User.role == Role.admin.value).first() is not None @router.get("/status") def setup_status(db: Session = Depends(get_db)): return {"setup_required": not _has_admin(db)} @router.post("/admin") def create_initial_admin( data: dict, resp: Response, db: Session = Depends(get_db), ): # The setup endpoint is only open until the first admin exists. if _has_admin(db): raise HTTPException(status_code=409, detail="setup already completed") email = (data.get("email") or "").lower().strip() display_name = (data.get("display_name") or "").strip() password = data.get("password") or "" if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", email): raise HTTPException(status_code=400, detail="valid email required") if len(password) < 8: raise HTTPException(status_code=400, detail="password too short (min 8)") if not display_name: display_name = email.split("@", 1)[0] if db.query(User).filter(User.email == email).first(): raise HTTPException(status_code=409, detail="email exists") user = User( email=email, password_hash=hash_password(password), role=Role.admin.value, display_name=display_name, ) db.add(user) db.commit() db.refresh(user) # Log the installer in immediately after successful setup. set_session(resp, make_session_value(user.id)) return {"ok": True, "id": user.id, "email": user.email}