486 lines
15 KiB
Python
486 lines
15 KiB
Python
"""HTMX-driven admin dashboard for users, roles, and thread standards."""
|
|
|
|
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 iso_thread as iso_thread_crud
|
|
from app.crud import role as role_crud
|
|
from app.crud import unc_thread as unc_thread_crud
|
|
from app.crud import user as user_crud
|
|
from app.deps import get_current_web_user
|
|
from app.models.unc_thread import UncSeries
|
|
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),
|
|
"permissions": await role_crud.list_permissions(db),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/roles/{role_id}/edit", response_class=HTMLResponse)
|
|
async def edit_role_row(
|
|
request: Request,
|
|
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 None:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/_role_row_edit.html",
|
|
{"r": role, "permissions": await role_crud.list_permissions(db)},
|
|
)
|
|
|
|
|
|
@router.get("/roles/{role_id}/row", response_class=HTMLResponse)
|
|
async def role_row(
|
|
request: Request,
|
|
role_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("roles:read")),
|
|
):
|
|
role = await role_crud.get(db, role_id)
|
|
if role is None:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
return templates.TemplateResponse(request, "admin/_role_row.html", {"r": role})
|
|
|
|
|
|
@router.post("/roles/{role_id}", response_class=HTMLResponse)
|
|
async def update_role(
|
|
request: Request,
|
|
role_id: int,
|
|
permission_ids: list[int] = Form(default=[]),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("roles:write")),
|
|
):
|
|
role = await role_crud.get(db, role_id)
|
|
if role is None:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
await role_crud.update(db, role, {"permission_ids": permission_ids})
|
|
return templates.TemplateResponse(request, "admin/_role_row.html", {"r": role})
|
|
|
|
|
|
@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)
|
|
|
|
|
|
# --- ISO Threads -------------------------------------------------------------
|
|
|
|
|
|
@router.get("/iso-threads", response_class=HTMLResponse)
|
|
async def iso_threads_page(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(web_require("iso_threads:read")),
|
|
):
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/iso_threads.html",
|
|
{"user": user, "threads": await iso_thread_crud.list_all(db)},
|
|
)
|
|
|
|
|
|
@router.post("/iso-threads", response_class=HTMLResponse)
|
|
async def create_iso_thread(
|
|
request: Request,
|
|
diameter_mm: float = Form(...),
|
|
pitch_mm: float = Form(...),
|
|
depth_mm: str = Form(""),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("iso_threads:write")),
|
|
):
|
|
await iso_thread_crud.create(
|
|
db,
|
|
diameter_mm=diameter_mm,
|
|
pitch_mm=pitch_mm,
|
|
depth_mm=float(depth_mm) if depth_mm.strip() else None,
|
|
)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/_iso_threads_table.html",
|
|
{"threads": await iso_thread_crud.list_all(db)},
|
|
)
|
|
|
|
|
|
@router.get("/iso-threads/{thread_id}/edit", response_class=HTMLResponse)
|
|
async def edit_iso_thread_row(
|
|
request: Request,
|
|
thread_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("iso_threads:write")),
|
|
):
|
|
thread = await iso_thread_crud.get(db, thread_id)
|
|
if thread is None:
|
|
raise HTTPException(status_code=404, detail="ISO thread not found")
|
|
return templates.TemplateResponse(
|
|
request, "admin/_iso_thread_row_edit.html", {"t": thread}
|
|
)
|
|
|
|
|
|
@router.get("/iso-threads/{thread_id}/row", response_class=HTMLResponse)
|
|
async def iso_thread_row(
|
|
request: Request,
|
|
thread_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("iso_threads:read")),
|
|
):
|
|
thread = await iso_thread_crud.get(db, thread_id)
|
|
if thread is None:
|
|
raise HTTPException(status_code=404, detail="ISO thread not found")
|
|
return templates.TemplateResponse(
|
|
request, "admin/_iso_thread_row.html", {"t": thread}
|
|
)
|
|
|
|
|
|
@router.post("/iso-threads/{thread_id}", response_class=HTMLResponse)
|
|
async def update_iso_thread(
|
|
request: Request,
|
|
thread_id: int,
|
|
diameter_mm: float = Form(...),
|
|
pitch_mm: float = Form(...),
|
|
depth_mm: str = Form(""),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("iso_threads:write")),
|
|
):
|
|
thread = await iso_thread_crud.get(db, thread_id)
|
|
if thread is None:
|
|
raise HTTPException(status_code=404, detail="ISO thread not found")
|
|
await iso_thread_crud.update(
|
|
db,
|
|
thread,
|
|
{
|
|
"diameter_mm": diameter_mm,
|
|
"pitch_mm": pitch_mm,
|
|
"depth_mm": float(depth_mm) if depth_mm.strip() else None,
|
|
},
|
|
)
|
|
return templates.TemplateResponse(
|
|
request, "admin/_iso_thread_row.html", {"t": thread}
|
|
)
|
|
|
|
|
|
@router.delete("/iso-threads/{thread_id}")
|
|
async def delete_iso_thread(
|
|
thread_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("iso_threads:delete")),
|
|
):
|
|
thread = await iso_thread_crud.get(db, thread_id)
|
|
if thread is not None:
|
|
await iso_thread_crud.delete(db, thread)
|
|
return Response(status_code=status.HTTP_200_OK)
|
|
|
|
|
|
# --- UNC Threads -------------------------------------------------------------
|
|
|
|
|
|
@router.get("/unc-threads", response_class=HTMLResponse)
|
|
async def unc_threads_page(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(web_require("unc_threads:read")),
|
|
):
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/unc_threads.html",
|
|
{
|
|
"user": user,
|
|
"threads": await unc_thread_crud.list_all(db),
|
|
"series_options": list(UncSeries),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/unc-threads", response_class=HTMLResponse)
|
|
async def create_unc_thread(
|
|
request: Request,
|
|
diameter_fraction: str = Form(...),
|
|
tpi: int = Form(...),
|
|
series: UncSeries = Form(...),
|
|
depth_in: str = Form(""),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("unc_threads:write")),
|
|
):
|
|
await unc_thread_crud.create(
|
|
db,
|
|
diameter_fraction=diameter_fraction,
|
|
tpi=tpi,
|
|
series=series,
|
|
depth_in=float(depth_in) if depth_in.strip() else None,
|
|
)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/_unc_threads_table.html",
|
|
{"threads": await unc_thread_crud.list_all(db)},
|
|
)
|
|
|
|
|
|
@router.get("/unc-threads/{thread_id}/edit", response_class=HTMLResponse)
|
|
async def edit_unc_thread_row(
|
|
request: Request,
|
|
thread_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("unc_threads:write")),
|
|
):
|
|
thread = await unc_thread_crud.get(db, thread_id)
|
|
if thread is None:
|
|
raise HTTPException(status_code=404, detail="UNC thread not found")
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/_unc_thread_row_edit.html",
|
|
{"t": thread, "series_options": list(UncSeries)},
|
|
)
|
|
|
|
|
|
@router.get("/unc-threads/{thread_id}/row", response_class=HTMLResponse)
|
|
async def unc_thread_row(
|
|
request: Request,
|
|
thread_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("unc_threads:read")),
|
|
):
|
|
thread = await unc_thread_crud.get(db, thread_id)
|
|
if thread is None:
|
|
raise HTTPException(status_code=404, detail="UNC thread not found")
|
|
return templates.TemplateResponse(
|
|
request, "admin/_unc_thread_row.html", {"t": thread}
|
|
)
|
|
|
|
|
|
@router.post("/unc-threads/{thread_id}", response_class=HTMLResponse)
|
|
async def update_unc_thread(
|
|
request: Request,
|
|
thread_id: int,
|
|
diameter_fraction: str = Form(...),
|
|
tpi: int = Form(...),
|
|
series: UncSeries = Form(...),
|
|
depth_in: str = Form(""),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("unc_threads:write")),
|
|
):
|
|
thread = await unc_thread_crud.get(db, thread_id)
|
|
if thread is None:
|
|
raise HTTPException(status_code=404, detail="UNC thread not found")
|
|
await unc_thread_crud.update(
|
|
db,
|
|
thread,
|
|
{
|
|
"diameter_fraction": diameter_fraction,
|
|
"tpi": tpi,
|
|
"series": series,
|
|
"depth_in": float(depth_in) if depth_in.strip() else None,
|
|
},
|
|
)
|
|
return templates.TemplateResponse(
|
|
request, "admin/_unc_thread_row.html", {"t": thread}
|
|
)
|
|
|
|
|
|
@router.delete("/unc-threads/{thread_id}")
|
|
async def delete_unc_thread(
|
|
thread_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(web_require("unc_threads:delete")),
|
|
):
|
|
thread = await unc_thread_crud.get(db, thread_id)
|
|
if thread is not None:
|
|
await unc_thread_crud.delete(db, thread)
|
|
return Response(status_code=status.HTTP_200_OK)
|