123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from auth import hash_password, require_admin_dep, require_auth
|
|
from db import get_db, row_to_dict
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["users"])
|
|
|
|
PUBLIC_FIELDS = "id, email, name, role, status, tier_id, created"
|
|
|
|
|
|
class CreateUserRequest(BaseModel):
|
|
email: str
|
|
name: str = Field(min_length=1)
|
|
password: str = Field(min_length=4)
|
|
role: str = "user"
|
|
|
|
|
|
class PatchUserRequest(BaseModel):
|
|
name: str | None = None
|
|
role: str | None = None
|
|
status: str | None = None
|
|
password: str | None = None
|
|
|
|
|
|
def _active_admin_count(conn, exclude_id: str | None = None) -> int:
|
|
if exclude_id:
|
|
row = conn.execute(
|
|
"SELECT COUNT(*) AS n FROM profiles WHERE role = 'admin' AND status = 'active' AND id != ?",
|
|
(exclude_id,),
|
|
).fetchone()
|
|
else:
|
|
row = conn.execute(
|
|
"SELECT COUNT(*) AS n FROM profiles WHERE role = 'admin' AND status = 'active'"
|
|
).fetchone()
|
|
return int(row["n"])
|
|
|
|
|
|
@router.get("")
|
|
def list_users(session: dict = Depends(require_admin_dep)):
|
|
with get_db() as conn:
|
|
rows = conn.execute(f"SELECT {PUBLIC_FIELDS} FROM profiles ORDER BY created").fetchall()
|
|
return [row_to_dict(row) for row in rows]
|
|
|
|
|
|
@router.post("")
|
|
def create_user(req: CreateUserRequest, session: dict = Depends(require_admin_dep)):
|
|
if req.role not in {"user", "admin"}:
|
|
raise HTTPException(400, "Rolle muss user oder admin sein")
|
|
profile_id = str(uuid.uuid4())
|
|
try:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO profiles (id, email, name, password_hash, role, status, tier_id)
|
|
VALUES (?, ?, ?, ?, ?, 'active', 'local')
|
|
""",
|
|
(
|
|
profile_id,
|
|
req.email.lower().strip(),
|
|
req.name.strip(),
|
|
hash_password(req.password),
|
|
req.role,
|
|
),
|
|
)
|
|
prof = row_to_dict(conn.execute(f"SELECT {PUBLIC_FIELDS} FROM profiles WHERE id = ?", (profile_id,)).fetchone())
|
|
except Exception as exc:
|
|
if "UNIQUE" in str(exc).upper():
|
|
raise HTTPException(409, "E-Mail ist bereits vergeben") from exc
|
|
raise
|
|
return prof
|
|
|
|
|
|
@router.patch("/{profile_id}")
|
|
def patch_user(profile_id: str, req: PatchUserRequest, session: dict = Depends(require_admin_dep)):
|
|
with get_db() as conn:
|
|
current = row_to_dict(conn.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)).fetchone())
|
|
if not current:
|
|
raise HTTPException(404, "Nutzer nicht gefunden")
|
|
name = req.name.strip() if req.name is not None else current["name"]
|
|
role = req.role if req.role is not None else current["role"]
|
|
status = req.status if req.status is not None else current["status"]
|
|
if role not in {"user", "admin"}:
|
|
raise HTTPException(400, "Rolle muss user oder admin sein")
|
|
if status not in {"active", "disabled"}:
|
|
raise HTTPException(400, "Status muss active oder disabled sein")
|
|
becomes_non_admin = current["role"] == "admin" and (role != "admin" or status == "disabled")
|
|
if becomes_non_admin and _active_admin_count(conn, exclude_id=profile_id) < 1:
|
|
raise HTTPException(400, "Der letzte aktive Admin kann nicht entfernt werden")
|
|
password_hash = current["password_hash"]
|
|
if req.password:
|
|
if len(req.password) < 4:
|
|
raise HTTPException(400, "Passwort muss mind. 4 Zeichen haben")
|
|
password_hash = hash_password(req.password)
|
|
conn.execute(
|
|
"""
|
|
UPDATE profiles SET name = ?, role = ?, status = ?, password_hash = ?
|
|
WHERE id = ?
|
|
""",
|
|
(name, role, status, password_hash, profile_id),
|
|
)
|
|
if status == "disabled":
|
|
conn.execute("DELETE FROM sessions WHERE profile_id = ?", (profile_id,))
|
|
return row_to_dict(conn.execute(f"SELECT {PUBLIC_FIELDS} FROM profiles WHERE id = ?", (profile_id,)).fetchone())
|
|
|
|
|
|
@router.get("/me")
|
|
def my_account(session: dict = Depends(require_auth)):
|
|
with get_db() as conn:
|
|
prof = row_to_dict(
|
|
conn.execute(
|
|
f"SELECT {PUBLIC_FIELDS} FROM profiles WHERE id = ?",
|
|
(session["profile_id"],),
|
|
).fetchone()
|
|
)
|
|
if not prof:
|
|
raise HTTPException(401, "Profil nicht gefunden")
|
|
return prof
|