59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
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}
|