113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
"""Management users admin (spec sections 7.1, 9)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from sqlalchemy import select
|
|
|
|
from app import audit
|
|
from app.deps import CurrentUser, DbSession, require_roles
|
|
from app.models import ManagementRole, ManagementUser
|
|
from app.schemas import (
|
|
ManagementUserCreate,
|
|
ManagementUserOut,
|
|
ManagementUserUpdate,
|
|
)
|
|
from app.security import hash_password
|
|
|
|
router = APIRouter(prefix="/api/v1/users", tags=["users"])
|
|
|
|
AdminUser = Annotated[CurrentUser, Depends(require_roles(ManagementRole.SUPER_ADMIN))]
|
|
|
|
|
|
@router.get("", response_model=list[ManagementUserOut])
|
|
def list_users(db: DbSession, _admin: AdminUser):
|
|
return db.scalars(select(ManagementUser)).all()
|
|
|
|
|
|
@router.post("", response_model=ManagementUserOut, status_code=status.HTTP_201_CREATED)
|
|
def create_user(
|
|
body: ManagementUserCreate,
|
|
request: Request,
|
|
db: DbSession,
|
|
_admin: AdminUser,
|
|
):
|
|
if db.scalar(select(ManagementUser).where(ManagementUser.username == body.username)):
|
|
raise HTTPException(status.HTTP_409_CONFLICT, "username already exists")
|
|
user = ManagementUser(
|
|
username=body.username,
|
|
password_hash=hash_password(body.password),
|
|
display_name=body.display_name,
|
|
role=body.role,
|
|
must_change_password=True,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
audit.record(
|
|
db,
|
|
"USER_CREATED",
|
|
request=request,
|
|
user=body.username,
|
|
object_type="management_user",
|
|
object_id=body.username,
|
|
new_value={"role": body.role.value},
|
|
)
|
|
return user
|
|
|
|
|
|
@router.patch("/{username}", response_model=ManagementUserOut)
|
|
def update_user(
|
|
username: str,
|
|
body: ManagementUserUpdate,
|
|
request: Request,
|
|
db: DbSession,
|
|
_admin: AdminUser,
|
|
):
|
|
user = db.scalar(select(ManagementUser).where(ManagementUser.username == username))
|
|
if user is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "user not found")
|
|
old = {"role": user.role.value, "enabled": user.enabled}
|
|
if body.display_name is not None:
|
|
user.display_name = body.display_name
|
|
if body.role is not None:
|
|
user.role = body.role
|
|
if body.enabled is not None:
|
|
user.enabled = body.enabled
|
|
db.flush()
|
|
audit.record(
|
|
db,
|
|
"USER_UPDATED",
|
|
request=request,
|
|
user=username,
|
|
object_type="management_user",
|
|
object_id=username,
|
|
old_value=old,
|
|
new_value={"role": user.role.value, "enabled": user.enabled},
|
|
)
|
|
return user
|
|
|
|
|
|
@router.delete("/{username}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_user(
|
|
username: str,
|
|
request: Request,
|
|
db: DbSession,
|
|
_admin: AdminUser,
|
|
):
|
|
if username == "admin":
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, "cannot delete bootstrap admin")
|
|
user = db.scalar(select(ManagementUser).where(ManagementUser.username == username))
|
|
if user is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "user not found")
|
|
db.delete(user)
|
|
audit.record(
|
|
db,
|
|
"USER_DELETED",
|
|
request=request,
|
|
user=username,
|
|
object_type="management_user",
|
|
object_id=username,
|
|
)
|
|
return None |