1358 lines
46 KiB
Python
1358 lines
46 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 audit as audit_crud
|
||
from app.crud import imported_part as imported_part_crud
|
||
from app.crud import imported_part_config as imported_part_config_crud
|
||
from app.crud import iso_thread as iso_thread_crud
|
||
from app.crud import manufacturer as manufacturer_crud
|
||
from app.crud import part_colour as part_colour_crud
|
||
from app.crud import part_type as part_type_crud
|
||
from app.crud import role as role_crud
|
||
from app.crud import scope_model as scope_model_crud
|
||
from app.crud import static_connection as static_connection_crud
|
||
from app.crud import sub_part_type as sub_part_type_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.services import odata as odata_service
|
||
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
|
||
|
||
|
||
async def _role_perm_ctx(db: AsyncSession) -> dict:
|
||
all_perms = await role_crud.list_permissions(db)
|
||
scopes = sorted({p.name.split(":")[0] for p in all_perms})
|
||
perm_map = {p.name: p for p in all_perms}
|
||
return {
|
||
"permissions": all_perms,
|
||
"scopes": scopes,
|
||
"perm_map": perm_map,
|
||
"actions": ["read", "write", "delete"],
|
||
}
|
||
|
||
|
||
@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),
|
||
actor: User = Depends(web_require("users:write")),
|
||
):
|
||
if not await user_crud.get_by_email(db, email):
|
||
created = await user_crud.create(
|
||
db,
|
||
email=email,
|
||
password=password,
|
||
full_name=full_name or None,
|
||
is_superuser=is_superuser,
|
||
role_ids=role_ids,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="User",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("User", created.id, created.email),
|
||
)
|
||
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),
|
||
actor: 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")
|
||
changes = {
|
||
"full_name": full_name or None,
|
||
"password": password or None,
|
||
"is_active": is_active,
|
||
"is_superuser": is_superuser,
|
||
"role_ids": role_ids,
|
||
}
|
||
await user_crud.update(db, target, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="User",
|
||
resource_id=user_id,
|
||
description=audit_crud.describe_update("User", user_id, changes),
|
||
)
|
||
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),
|
||
actor: User = Depends(web_require("users:delete")),
|
||
):
|
||
target = await user_crud.get(db, user_id)
|
||
if target is not None:
|
||
email = target.email
|
||
await user_crud.delete(db, target)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="User",
|
||
resource_id=user_id,
|
||
description=audit_crud.describe_delete("User", user_id, email),
|
||
)
|
||
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), **(await _role_perm_ctx(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),
|
||
actor: User = Depends(web_require("roles:write")),
|
||
):
|
||
if not await role_crud.get_by_name(db, name):
|
||
created = await role_crud.create(
|
||
db,
|
||
name=name,
|
||
description=description or None,
|
||
permission_ids=permission_ids,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="Role",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("Role", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_roles_table.html",
|
||
{"roles": await role_crud.list_roles(db), **(await _role_perm_ctx(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, **(await _role_perm_ctx(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, **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@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),
|
||
actor: 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")
|
||
changes = {"permission_ids": permission_ids}
|
||
await role_crud.update(db, role, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="Role",
|
||
resource_id=role_id,
|
||
description=audit_crud.describe_update("Role", role_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_role_row.html",
|
||
{"r": role, **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@router.delete("/roles/{role_id}")
|
||
async def delete_role(
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("roles:write")),
|
||
):
|
||
role = await role_crud.get(db, role_id)
|
||
if role is not None:
|
||
name = role.name
|
||
await role_crud.delete(db, role)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="Role",
|
||
resource_id=role_id,
|
||
description=audit_crud.describe_delete("Role", role_id, name),
|
||
)
|
||
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),
|
||
actor: User = Depends(web_require("iso_threads:write")),
|
||
):
|
||
thread = 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,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="IsoThread",
|
||
resource_id=thread.id,
|
||
description=audit_crud.describe_create("IsoThread", thread.id, f"M{thread.diameter_mm}×{thread.pitch_mm}"),
|
||
)
|
||
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),
|
||
actor: 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")
|
||
changes = {
|
||
"diameter_mm": diameter_mm,
|
||
"pitch_mm": pitch_mm,
|
||
"depth_mm": float(depth_mm) if depth_mm.strip() else None,
|
||
}
|
||
await iso_thread_crud.update(db, thread, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="IsoThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_update("IsoThread", thread_id, changes),
|
||
)
|
||
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),
|
||
actor: User = Depends(web_require("iso_threads:delete")),
|
||
):
|
||
thread = await iso_thread_crud.get(db, thread_id)
|
||
if thread is not None:
|
||
name = f"M{thread.diameter_mm}×{thread.pitch_mm}"
|
||
await iso_thread_crud.delete(db, thread)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="IsoThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_delete("IsoThread", thread_id, name),
|
||
)
|
||
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),
|
||
actor: User = Depends(web_require("unc_threads:write")),
|
||
):
|
||
thread = 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,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="UncThread",
|
||
resource_id=thread.id,
|
||
description=audit_crud.describe_create("UncThread", thread.id, f"{thread.diameter_fraction}-{thread.tpi} {thread.series.value}"),
|
||
)
|
||
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),
|
||
actor: 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")
|
||
changes = {
|
||
"diameter_fraction": diameter_fraction,
|
||
"tpi": tpi,
|
||
"series": series,
|
||
"depth_in": float(depth_in) if depth_in.strip() else None,
|
||
}
|
||
await unc_thread_crud.update(db, thread, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="UncThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_update("UncThread", thread_id, changes),
|
||
)
|
||
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),
|
||
actor: User = Depends(web_require("unc_threads:delete")),
|
||
):
|
||
thread = await unc_thread_crud.get(db, thread_id)
|
||
if thread is not None:
|
||
name = f"{thread.diameter_fraction}-{thread.tpi} {thread.series.value}"
|
||
await unc_thread_crud.delete(db, thread)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="UncThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_delete("UncThread", thread_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Manufacturers -----------------------------------------------------------
|
||
|
||
|
||
@router.get("/manufacturers", response_class=HTMLResponse)
|
||
async def manufacturers_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("manufacturers:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/manufacturers.html",
|
||
{"user": user, "manufacturers": await manufacturer_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/manufacturers", response_class=HTMLResponse)
|
||
async def create_manufacturer(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
url: str = Form(...),
|
||
logo_url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("manufacturers:write")),
|
||
):
|
||
created = await manufacturer_crud.create(
|
||
db,
|
||
name=name,
|
||
url=url,
|
||
logo_url=logo_url.strip() or None,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="Manufacturer",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("Manufacturer", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_manufacturers_table.html",
|
||
{"manufacturers": await manufacturer_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/manufacturers/{manufacturer_id}/edit", response_class=HTMLResponse)
|
||
async def edit_manufacturer_row(
|
||
request: Request,
|
||
manufacturer_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("manufacturers:write")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is None:
|
||
raise HTTPException(status_code=404, detail="Manufacturer not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_manufacturer_row_edit.html", {"m": m}
|
||
)
|
||
|
||
|
||
@router.get("/manufacturers/{manufacturer_id}/row", response_class=HTMLResponse)
|
||
async def manufacturer_row(
|
||
request: Request,
|
||
manufacturer_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("manufacturers:read")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is None:
|
||
raise HTTPException(status_code=404, detail="Manufacturer not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_manufacturer_row.html", {"m": m}
|
||
)
|
||
|
||
|
||
@router.post("/manufacturers/{manufacturer_id}", response_class=HTMLResponse)
|
||
async def update_manufacturer(
|
||
request: Request,
|
||
manufacturer_id: int,
|
||
name: str = Form(...),
|
||
url: str = Form(...),
|
||
logo_url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("manufacturers:write")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is None:
|
||
raise HTTPException(status_code=404, detail="Manufacturer not found")
|
||
changes = {"name": name, "url": url, "logo_url": logo_url.strip() or None}
|
||
await manufacturer_crud.update(db, m, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="Manufacturer",
|
||
resource_id=manufacturer_id,
|
||
description=audit_crud.describe_update("Manufacturer", manufacturer_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_manufacturer_row.html", {"m": m}
|
||
)
|
||
|
||
|
||
@router.delete("/manufacturers/{manufacturer_id}")
|
||
async def delete_manufacturer(
|
||
manufacturer_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("manufacturers:delete")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is not None:
|
||
name = m.name
|
||
await manufacturer_crud.delete(db, m)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="Manufacturer",
|
||
resource_id=manufacturer_id,
|
||
description=audit_crud.describe_delete("Manufacturer", manufacturer_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Scope Models ------------------------------------------------------------
|
||
|
||
|
||
@router.get("/scope-models", response_class=HTMLResponse)
|
||
async def scope_models_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("scope_models:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/scope_models.html",
|
||
{
|
||
"user": user,
|
||
"scope_models": await scope_model_crud.list_all(db),
|
||
"manufacturers": await manufacturer_crud.list_all(db),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/scope-models", response_class=HTMLResponse)
|
||
async def create_scope_model(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
manufacturer_id: int = Form(...),
|
||
url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("scope_models:write")),
|
||
):
|
||
created = await scope_model_crud.create(
|
||
db,
|
||
name=name,
|
||
manufacturer_id=manufacturer_id,
|
||
url=url.strip() or None,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="ScopeModel",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("ScopeModel", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_models_table.html",
|
||
{"scope_models": await scope_model_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/scope-models/{scope_model_id}/edit", response_class=HTMLResponse)
|
||
async def edit_scope_model_row(
|
||
request: Request,
|
||
scope_model_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("scope_models:write")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is None:
|
||
raise HTTPException(status_code=404, detail="Scope model not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_model_row_edit.html",
|
||
{"sm": sm, "manufacturers": await manufacturer_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/scope-models/{scope_model_id}/row", response_class=HTMLResponse)
|
||
async def scope_model_row(
|
||
request: Request,
|
||
scope_model_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("scope_models:read")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is None:
|
||
raise HTTPException(status_code=404, detail="Scope model not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_scope_model_row.html", {"sm": sm}
|
||
)
|
||
|
||
|
||
@router.post("/scope-models/{scope_model_id}", response_class=HTMLResponse)
|
||
async def update_scope_model(
|
||
request: Request,
|
||
scope_model_id: int,
|
||
name: str = Form(...),
|
||
manufacturer_id: int = Form(...),
|
||
url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("scope_models:write")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is None:
|
||
raise HTTPException(status_code=404, detail="Scope model not found")
|
||
changes = {"name": name, "manufacturer_id": manufacturer_id, "url": url.strip() or None}
|
||
await scope_model_crud.update(db, sm, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="ScopeModel",
|
||
resource_id=scope_model_id,
|
||
description=audit_crud.describe_update("ScopeModel", scope_model_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_scope_model_row.html", {"sm": sm}
|
||
)
|
||
|
||
|
||
@router.delete("/scope-models/{scope_model_id}")
|
||
async def delete_scope_model(
|
||
scope_model_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("scope_models:delete")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is not None:
|
||
name = sm.name
|
||
await scope_model_crud.delete(db, sm)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="ScopeModel",
|
||
resource_id=scope_model_id,
|
||
description=audit_crud.describe_delete("ScopeModel", scope_model_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Part Types --------------------------------------------------------------
|
||
|
||
|
||
@router.get("/part-types", response_class=HTMLResponse)
|
||
async def part_types_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("part_types:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/part_types.html",
|
||
{"user": user, "part_types": await part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/part-types", response_class=HTMLResponse)
|
||
async def create_part_type(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_types:write")),
|
||
):
|
||
if not await part_type_crud.get_by_name(db, name):
|
||
created = await part_type_crud.create(db, name=name)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="PartType",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("PartType", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_part_types_table.html",
|
||
{"part_types": await part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/part-types/{part_type_id}/edit", response_class=HTMLResponse)
|
||
async def edit_part_type_row(
|
||
request: Request,
|
||
part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_types:write")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is None:
|
||
raise HTTPException(status_code=404, detail="Part type not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_type_row_edit.html", {"pt": pt}
|
||
)
|
||
|
||
|
||
@router.get("/part-types/{part_type_id}/row", response_class=HTMLResponse)
|
||
async def part_type_row(
|
||
request: Request,
|
||
part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_types:read")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is None:
|
||
raise HTTPException(status_code=404, detail="Part type not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_type_row.html", {"pt": pt}
|
||
)
|
||
|
||
|
||
@router.post("/part-types/{part_type_id}", response_class=HTMLResponse)
|
||
async def update_part_type(
|
||
request: Request,
|
||
part_type_id: int,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_types:write")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is None:
|
||
raise HTTPException(status_code=404, detail="Part type not found")
|
||
changes = {"name": name}
|
||
await part_type_crud.update(db, pt, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="PartType",
|
||
resource_id=part_type_id,
|
||
description=audit_crud.describe_update("PartType", part_type_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_type_row.html", {"pt": pt}
|
||
)
|
||
|
||
|
||
@router.delete("/part-types/{part_type_id}")
|
||
async def delete_part_type(
|
||
part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_types:delete")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is not None:
|
||
name = pt.name
|
||
await part_type_crud.delete(db, pt)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="PartType",
|
||
resource_id=part_type_id,
|
||
description=audit_crud.describe_delete("PartType", part_type_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Sub Part Types ----------------------------------------------------------
|
||
|
||
|
||
@router.get("/sub-part-types", response_class=HTMLResponse)
|
||
async def sub_part_types_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("sub_part_types:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/sub_part_types.html",
|
||
{
|
||
"user": user,
|
||
"sub_part_types": await sub_part_type_crud.list_all(db),
|
||
"part_types": await part_type_crud.list_all(db),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/sub-part-types", response_class=HTMLResponse)
|
||
async def create_sub_part_type(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
part_type_id: int = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("sub_part_types:write")),
|
||
):
|
||
if not await sub_part_type_crud.get_by_name_and_part_type(db, name, part_type_id):
|
||
created = await sub_part_type_crud.create(db, name=name, part_type_id=part_type_id)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="SubPartType",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("SubPartType", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_sub_part_types_table.html",
|
||
{"sub_part_types": await sub_part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/sub-part-types/{sub_part_type_id}/edit", response_class=HTMLResponse)
|
||
async def edit_sub_part_type_row(
|
||
request: Request,
|
||
sub_part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("sub_part_types:write")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is None:
|
||
raise HTTPException(status_code=404, detail="Sub part type not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_sub_part_type_row_edit.html",
|
||
{"spt": spt, "part_types": await part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/sub-part-types/{sub_part_type_id}/row", response_class=HTMLResponse)
|
||
async def sub_part_type_row(
|
||
request: Request,
|
||
sub_part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("sub_part_types:read")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is None:
|
||
raise HTTPException(status_code=404, detail="Sub part type not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_sub_part_type_row.html", {"spt": spt}
|
||
)
|
||
|
||
|
||
@router.post("/sub-part-types/{sub_part_type_id}", response_class=HTMLResponse)
|
||
async def update_sub_part_type(
|
||
request: Request,
|
||
sub_part_type_id: int,
|
||
name: str = Form(...),
|
||
part_type_id: int = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("sub_part_types:write")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is None:
|
||
raise HTTPException(status_code=404, detail="Sub part type not found")
|
||
changes = {"name": name, "part_type_id": part_type_id}
|
||
await sub_part_type_crud.update(db, spt, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="SubPartType",
|
||
resource_id=sub_part_type_id,
|
||
description=audit_crud.describe_update("SubPartType", sub_part_type_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_sub_part_type_row.html", {"spt": spt}
|
||
)
|
||
|
||
|
||
@router.delete("/sub-part-types/{sub_part_type_id}")
|
||
async def delete_sub_part_type(
|
||
sub_part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("sub_part_types:delete")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is not None:
|
||
name = spt.name
|
||
await sub_part_type_crud.delete(db, spt)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="SubPartType",
|
||
resource_id=sub_part_type_id,
|
||
description=audit_crud.describe_delete("SubPartType", sub_part_type_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Imported Parts ----------------------------------------------------------
|
||
|
||
|
||
@router.get("/imported-parts", response_class=HTMLResponse)
|
||
async def imported_parts_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("imported_parts:read")),
|
||
):
|
||
config = await imported_part_config_crud.get(db)
|
||
config_view = {
|
||
"odata_url": config.odata_url if config else None,
|
||
"username": config.username if config else None,
|
||
"has_password": bool(config and config.password),
|
||
}
|
||
parts = await imported_part_crud.list_all(db)
|
||
total = await imported_part_crud.count(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/imported_parts.html",
|
||
{"user": user, "config": config_view, "parts": parts, "total": total},
|
||
)
|
||
|
||
|
||
@router.post("/imported-parts/config", response_class=HTMLResponse)
|
||
async def save_imported_part_config(
|
||
request: Request,
|
||
odata_url: str = Form(""),
|
||
username: str = Form(""),
|
||
password: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("imported_parts:write")),
|
||
):
|
||
current = await imported_part_config_crud.get(db)
|
||
# Keep existing password when field submitted blank
|
||
kept_password = password.strip() or (current.password if current else None)
|
||
await imported_part_config_crud.upsert(
|
||
db,
|
||
odata_url=odata_url.strip() or None,
|
||
username=username.strip() or None,
|
||
password=kept_password,
|
||
)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="ImportedPartConfig",
|
||
resource_id=1,
|
||
description="Updated ImportedPart OData configuration",
|
||
)
|
||
return HTMLResponse("Configuration saved.", status_code=200)
|
||
|
||
|
||
@router.post("/imported-parts/refresh", response_class=HTMLResponse)
|
||
async def refresh_imported_parts(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("imported_parts:write")),
|
||
):
|
||
config = await imported_part_config_crud.get(db)
|
||
if not config or not config.odata_url:
|
||
parts = await imported_part_crud.list_all(db)
|
||
total = await imported_part_crud.count(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table.html",
|
||
{"parts": parts, "total": total, "sync_error": "OData URL is not configured."},
|
||
)
|
||
try:
|
||
records = await odata_service.fetch(config.odata_url, config.username, config.password)
|
||
created, updated = await imported_part_crud.upsert_bulk(db, records)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="ImportedPart",
|
||
resource_id=None,
|
||
description=f"OData sync: {created} created, {updated} updated ({len(records)} total records)",
|
||
)
|
||
parts = await imported_part_crud.list_all(db)
|
||
total = await imported_part_crud.count(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table.html",
|
||
{
|
||
"parts": parts,
|
||
"total": total,
|
||
"sync_result": f"Sync complete: {created} created, {updated} updated.",
|
||
},
|
||
)
|
||
except Exception as exc:
|
||
parts = await imported_part_crud.list_all(db)
|
||
total = await imported_part_crud.count(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table.html",
|
||
{"parts": parts, "total": total, "sync_error": f"Sync failed: {exc}"},
|
||
)
|
||
|
||
|
||
# --- Part Colours ------------------------------------------------------------
|
||
|
||
|
||
@router.get("/part-colours", response_class=HTMLResponse)
|
||
async def part_colours_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("part_colours:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/part_colours.html",
|
||
{"user": user, "part_colours": await part_colour_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/part-colours", response_class=HTMLResponse)
|
||
async def create_part_colour(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_colours:write")),
|
||
):
|
||
if not await part_colour_crud.get_by_name(db, name):
|
||
created = await part_colour_crud.create(db, name=name)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="PartColour",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("PartColour", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_part_colours_table.html",
|
||
{"part_colours": await part_colour_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/part-colours/{part_colour_id}/edit", response_class=HTMLResponse)
|
||
async def edit_part_colour_row(
|
||
request: Request,
|
||
part_colour_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_colours:write")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is None:
|
||
raise HTTPException(status_code=404, detail="Part colour not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_colour_row_edit.html", {"pc": pc}
|
||
)
|
||
|
||
|
||
@router.get("/part-colours/{part_colour_id}/row", response_class=HTMLResponse)
|
||
async def part_colour_row(
|
||
request: Request,
|
||
part_colour_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_colours:read")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is None:
|
||
raise HTTPException(status_code=404, detail="Part colour not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_colour_row.html", {"pc": pc}
|
||
)
|
||
|
||
|
||
@router.post("/part-colours/{part_colour_id}", response_class=HTMLResponse)
|
||
async def update_part_colour(
|
||
request: Request,
|
||
part_colour_id: int,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_colours:write")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is None:
|
||
raise HTTPException(status_code=404, detail="Part colour not found")
|
||
changes = {"name": name}
|
||
await part_colour_crud.update(db, pc, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="PartColour",
|
||
resource_id=part_colour_id,
|
||
description=audit_crud.describe_update("PartColour", part_colour_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_colour_row.html", {"pc": pc}
|
||
)
|
||
|
||
|
||
@router.delete("/part-colours/{part_colour_id}")
|
||
async def delete_part_colour(
|
||
part_colour_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_colours:delete")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is not None:
|
||
name = pc.name
|
||
await part_colour_crud.delete(db, pc)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="PartColour",
|
||
resource_id=part_colour_id,
|
||
description=audit_crud.describe_delete("PartColour", part_colour_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Static Connections -------------------------------------------------------
|
||
|
||
|
||
@router.get("/static-connections", response_class=HTMLResponse)
|
||
async def static_connections_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("static_connections:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/static_connections.html",
|
||
{"user": user, "static_connections": await static_connection_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/static-connections", response_class=HTMLResponse)
|
||
async def create_static_connection(
|
||
request: Request,
|
||
od_mm: float = Form(...),
|
||
od_in: float = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("static_connections:write")),
|
||
):
|
||
if not await static_connection_crud.get_by_od_mm(db, od_mm):
|
||
created = await static_connection_crud.create(db, od_mm=od_mm, od_in=od_in)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="create", resource_type="StaticConnection",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"StaticConnection", created.id, f"{created.od_mm} mm"
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_static_connections_table.html",
|
||
{"static_connections": await static_connection_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/static-connections/{static_connection_id}/edit", response_class=HTMLResponse)
|
||
async def edit_static_connection_row(
|
||
request: Request,
|
||
static_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("static_connections:write")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is None:
|
||
raise HTTPException(status_code=404, detail="Static connection not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_static_connection_row_edit.html", {"sc": sc}
|
||
)
|
||
|
||
|
||
@router.get("/static-connections/{static_connection_id}/row", response_class=HTMLResponse)
|
||
async def static_connection_row(
|
||
request: Request,
|
||
static_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("static_connections:read")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is None:
|
||
raise HTTPException(status_code=404, detail="Static connection not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_static_connection_row.html", {"sc": sc}
|
||
)
|
||
|
||
|
||
@router.post("/static-connections/{static_connection_id}", response_class=HTMLResponse)
|
||
async def update_static_connection(
|
||
request: Request,
|
||
static_connection_id: int,
|
||
od_mm: float = Form(...),
|
||
od_in: float = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("static_connections:write")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is None:
|
||
raise HTTPException(status_code=404, detail="Static connection not found")
|
||
changes = {"od_mm": od_mm, "od_in": od_in}
|
||
await static_connection_crud.update(db, sc, changes)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="update", resource_type="StaticConnection",
|
||
resource_id=static_connection_id,
|
||
description=audit_crud.describe_update("StaticConnection", static_connection_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_static_connection_row.html", {"sc": sc}
|
||
)
|
||
|
||
|
||
@router.delete("/static-connections/{static_connection_id}")
|
||
async def delete_static_connection(
|
||
static_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("static_connections:delete")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is not None:
|
||
od_mm = sc.od_mm
|
||
await static_connection_crud.delete(db, sc)
|
||
await audit_crud.log(
|
||
db, actor=actor, action="delete", resource_type="StaticConnection",
|
||
resource_id=static_connection_id,
|
||
description=audit_crud.describe_delete(
|
||
"StaticConnection", static_connection_id, f"{od_mm} mm"
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Audit Log ---------------------------------------------------------------
|
||
|
||
|
||
@router.get("/audit-log", response_class=HTMLResponse)
|
||
async def audit_log_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("audit_log:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/audit_log.html",
|
||
{"user": user, "entries": await audit_crud.list_recent(db)},
|
||
)
|