skeleton-api/app/web/admin.py

"""HTMX-driven admin dashboard for users and roles."""
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.crud import role as role_crud
from app.crud import user as user_crud
from app.deps import get_current_web_user
from app.models.user import User
from app.web.templates import templates

router = APIRouter(prefix="/admin", include_in_schema=False)


def web_require(*perms: str):
    """Web dependency: allow superusers or holders of all given permissions."""

    async def checker(user: User = Depends(get_current_web_user)) -> User:
        if user.is_superuser or set(perms).issubset(user.permissions):
            return user
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not have access to this page.",
        )

    return checker


@router.get("", response_class=HTMLResponse)
async def admin_home():
    return RedirectResponse(url="/admin/users", status_code=status.HTTP_302_FOUND)


# --- Users -------------------------------------------------------------------

@router.get("/users", response_class=HTMLResponse)
async def users_page(
    request: Request,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(web_require("users:read")),
):
    return templates.TemplateResponse(
        request,
        "admin/users.html",
        {
            "user": user,
            "users": await user_crud.list_users(db),
            "roles": await role_crud.list_roles(db),
        },
    )


@router.post("/users", response_class=HTMLResponse)
async def create_user(
    request: Request,
    email: str = Form(...),
    password: str = Form(...),
    full_name: str = Form(""),
    is_superuser: bool = Form(False),
    role_ids: list[int] = Form(default=[]),
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("users:write")),
):
    if not await user_crud.get_by_email(db, email):
        await user_crud.create(
            db,
            email=email,
            password=password,
            full_name=full_name or None,
            is_superuser=is_superuser,
            role_ids=role_ids,
        )
    return templates.TemplateResponse(
        request,
        "admin/_users_table.html",
        {"users": await user_crud.list_users(db), "roles": await role_crud.list_roles(db)},
    )


@router.get("/users/{user_id}/edit", response_class=HTMLResponse)
async def edit_user_row(
    request: Request,
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("users:write")),
):
    target = await user_crud.get(db, user_id)
    if target is None:
        raise HTTPException(status_code=404, detail="User not found")
    return templates.TemplateResponse(
        request,
        "admin/_user_row_edit.html",
        {"u": target, "roles": await role_crud.list_roles(db)},
    )


@router.get("/users/{user_id}/row", response_class=HTMLResponse)
async def user_row(
    request: Request,
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("users:read")),
):
    target = await user_crud.get(db, user_id)
    if target is None:
        raise HTTPException(status_code=404, detail="User not found")
    return templates.TemplateResponse(request, "admin/_user_row.html", {"u": target})


@router.post("/users/{user_id}", response_class=HTMLResponse)
async def update_user(
    request: Request,
    user_id: int,
    full_name: str = Form(""),
    password: str = Form(""),
    is_active: bool = Form(False),
    is_superuser: bool = Form(False),
    role_ids: list[int] = Form(default=[]),
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("users:write")),
):
    target = await user_crud.get(db, user_id)
    if target is None:
        raise HTTPException(status_code=404, detail="User not found")
    await user_crud.update(
        db,
        target,
        {
            "full_name": full_name or None,
            "password": password or None,
            "is_active": is_active,
            "is_superuser": is_superuser,
            "role_ids": role_ids,
        },
    )
    return templates.TemplateResponse(request, "admin/_user_row.html", {"u": target})


@router.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("users:delete")),
):
    target = await user_crud.get(db, user_id)
    if target is not None:
        await user_crud.delete(db, target)
    # Empty body — HTMX swaps the row out of existence.
    return Response(status_code=status.HTTP_200_OK)


# --- Roles -------------------------------------------------------------------

@router.get("/roles", response_class=HTMLResponse)
async def roles_page(
    request: Request,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(web_require("roles:read")),
):
    return templates.TemplateResponse(
        request,
        "admin/roles.html",
        {
            "user": user,
            "roles": await role_crud.list_roles(db),
            "permissions": await role_crud.list_permissions(db),
        },
    )


@router.post("/roles", response_class=HTMLResponse)
async def create_role(
    request: Request,
    name: str = Form(...),
    description: str = Form(""),
    permission_ids: list[int] = Form(default=[]),
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("roles:write")),
):
    if not await role_crud.get_by_name(db, name):
        await role_crud.create(
            db,
            name=name,
            description=description or None,
            permission_ids=permission_ids,
        )
    return templates.TemplateResponse(
        request,
        "admin/_roles_table.html",
        {"roles": await role_crud.list_roles(db)},
    )


@router.delete("/roles/{role_id}")
async def delete_role(
    role_id: int,
    db: AsyncSession = Depends(get_db),
    _: User = Depends(web_require("roles:write")),
):
    role = await role_crud.get(db, role_id)
    if role is not None:
        await role_crud.delete(db, role)
    return Response(status_code=status.HTTP_200_OK)