anti
741e6bb0d3
- Rename project to stealergram throughout - Add pyproject.toml (replaces requirements.txt split, folds pytest.ini) - Replace all em-dashes with hyphens across all source files Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""
|
|
web/routes/users.py - User CRUD (superadmin only).
|
|
|
|
GET /users → list all users
|
|
POST /users → create a new user
|
|
PATCH /users/{id} → update role / password / active flag
|
|
DELETE /users/{id} → deactivate (cannot delete self or other superadmins)
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from web import db
|
|
from web.dependencies import require_role
|
|
from web.models import CreateUserRequest, UpdateUserRequest
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _templates(request: Request):
|
|
return request.app.state.templates
|
|
|
|
|
|
@router.get("/users")
|
|
async def list_users(request: Request, user=Depends(require_role("superadmin"))):
|
|
users = db.list_users()
|
|
return _templates(request).TemplateResponse(
|
|
request, "users.html",
|
|
{"user": dict(user), "users": [dict(u) for u in users]},
|
|
)
|
|
|
|
|
|
@router.post("/users")
|
|
async def create_user(
|
|
payload: CreateUserRequest,
|
|
_user=Depends(require_role("superadmin")),
|
|
):
|
|
try:
|
|
user_id = db.create_user(payload.username, payload.password, payload.role)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
|
|
return {"id": user_id}
|
|
|
|
|
|
@router.patch("/users/{user_id}")
|
|
async def update_user(
|
|
user_id: str,
|
|
payload: UpdateUserRequest,
|
|
acting_user=Depends(require_role("superadmin")),
|
|
):
|
|
target = db.get_user_by_id(user_id)
|
|
if target is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Cannot demote another superadmin via role patch
|
|
if target["role"] == "superadmin" and payload.role and payload.role != "superadmin":
|
|
raise HTTPException(status_code=403, detail="Cannot demote another superadmin")
|
|
|
|
updates: dict = {}
|
|
if payload.password is not None:
|
|
updates["password"] = payload.password
|
|
if payload.role is not None:
|
|
updates["role"] = payload.role
|
|
if payload.is_active is not None:
|
|
updates["is_active"] = 1 if payload.is_active else 0
|
|
|
|
db.update_user(user_id, **updates)
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.delete("/users/{user_id}")
|
|
async def deactivate_user(
|
|
user_id: str,
|
|
acting_user=Depends(require_role("superadmin")),
|
|
):
|
|
if user_id == acting_user["id"]:
|
|
raise HTTPException(status_code=403, detail="Cannot deactivate yourself")
|
|
target = db.get_user_by_id(user_id)
|
|
if target is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
db.deactivate_user(user_id)
|
|
return {"status": "ok"}
|