234 lines
7.6 KiB
Python
234 lines
7.6 KiB
Python
"""Top-level, read-only scope fit guide at ``/scopes``.
|
|
|
|
The site's primary view: a table of scopes with their specs, plus one column per
|
|
non-Scope part type listing the accessories that fit each scope. Supports HTMX
|
|
search/filter over scope attributes and required accessory fits.
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
|
|
from fastapi.responses import HTMLResponse
|
|
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 augmented_part as augmented_part_crud
|
|
from app.crud import manufacturer as manufacturer_crud
|
|
from app.crud import part_type as part_type_crud
|
|
from app.crud import scope_model as scope_model_crud
|
|
from app.crud import sub_part_type as sub_part_type_crud
|
|
from app.deps import get_optional_web_user
|
|
from app.models.user import User
|
|
from app.services import scope_guide as scope_guide_service
|
|
from app.web.admin import web_require
|
|
from app.web.pagination import paginate
|
|
from app.web.templates import templates
|
|
|
|
router = APIRouter(include_in_schema=False)
|
|
|
|
_PAGE_SIZE = 50
|
|
_TABLE_TARGET = "#scopes-table-card"
|
|
_FILTER_PATH = "/scopes/filter"
|
|
_REQUIRED_PERMS = {"augmented_parts:read", "fits:read"}
|
|
|
|
|
|
def _to_int(value: str) -> int | None:
|
|
return int(value) if value.strip() else None
|
|
|
|
|
|
def _can_write_scopes(user: User | None) -> bool:
|
|
return user is not None and (
|
|
user.is_superuser or "augmented_parts:write" in user.permissions
|
|
)
|
|
|
|
|
|
async def _optional_scopes_user(
|
|
user: User | None = Depends(get_optional_web_user),
|
|
) -> User | None:
|
|
"""Anyone may view the scopes page; logged-in users still need the read perms.
|
|
|
|
Anonymous visitors are allowed through (and get an exact-fits-only view —
|
|
see ``only_exact`` in ``scope_guide.build``); a logged-in user missing the
|
|
permissions gets the usual 403, matching ``web_require``'s behaviour.
|
|
"""
|
|
if user is not None and not (
|
|
user.is_superuser or _REQUIRED_PERMS.issubset(user.permissions)
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="You do not have access to this page.",
|
|
)
|
|
return user
|
|
|
|
|
|
async def _build(db: AsyncSession, filters: dict, only_exact: bool = False) -> dict:
|
|
"""Run the guide query for the given raw (string) filter values.
|
|
|
|
Always returns the full matching row set — pagination is applied by the
|
|
caller by slicing ``rows`` (see ``_paginated_guide``).
|
|
"""
|
|
return await scope_guide_service.build(
|
|
db,
|
|
description=filters["description"],
|
|
manufacturer_id=_to_int(filters["manufacturer_id"]),
|
|
scope_model_id=_to_int(filters["scope_model_id"]),
|
|
sub_part_type_id=_to_int(filters["sub_part_type_id"]),
|
|
fits_part_type_id=_to_int(filters["fits_part_type_id"]),
|
|
only_exact=only_exact,
|
|
)
|
|
|
|
|
|
async def _paginated_guide(
|
|
db: AsyncSession,
|
|
filters: dict,
|
|
page: int,
|
|
page_size: int,
|
|
only_exact: bool = False,
|
|
) -> tuple[dict, dict]:
|
|
"""Run the guide query, then slice its rows to the requested page.
|
|
|
|
Returns ``(guide, pagination)`` where ``guide["rows"]`` is already sliced.
|
|
"""
|
|
guide = await _build(db, filters, only_exact=only_exact)
|
|
all_rows = guide["rows"]
|
|
pagination = paginate(
|
|
_FILTER_PATH, filters, page, page_size, len(all_rows), _TABLE_TARGET
|
|
)
|
|
guide["rows"] = all_rows[pagination["skip"] : pagination["skip"] + page_size]
|
|
return guide, pagination
|
|
|
|
|
|
async def _filter_options(db: AsyncSession) -> dict:
|
|
"""Dropdown option lists for the filter bar."""
|
|
scope_type = await part_type_crud.get_by_name(
|
|
db, scope_guide_service.SCOPE_TYPE_NAME
|
|
)
|
|
sub_part_types = [
|
|
spt
|
|
for spt in await sub_part_type_crud.list_all(db, limit=1000)
|
|
if scope_type is not None and spt.part_type_id == scope_type.id
|
|
]
|
|
return {
|
|
"manufacturers": await manufacturer_crud.list_all(db, limit=1000),
|
|
"scope_models": await scope_model_crud.list_all(db, limit=1000),
|
|
"sub_part_types": sub_part_types,
|
|
}
|
|
|
|
|
|
@router.get("/scopes", response_class=HTMLResponse)
|
|
async def scopes_page(
|
|
request: Request,
|
|
description: str = "",
|
|
manufacturer_id: str = "",
|
|
scope_model_id: str = "",
|
|
sub_part_type_id: str = "",
|
|
fits_part_type_id: str = "",
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User | None = Depends(_optional_scopes_user),
|
|
):
|
|
filters = {
|
|
"description": description,
|
|
"manufacturer_id": manufacturer_id,
|
|
"scope_model_id": scope_model_id,
|
|
"sub_part_type_id": sub_part_type_id,
|
|
"fits_part_type_id": fits_part_type_id,
|
|
}
|
|
only_exact = user is None
|
|
guide, pagination = await _paginated_guide(
|
|
db, filters, 1, _PAGE_SIZE, only_exact=only_exact
|
|
)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"scopes.html",
|
|
{
|
|
"user": user,
|
|
"filters": filters,
|
|
"can_write_scopes": _can_write_scopes(user),
|
|
"only_exact": only_exact,
|
|
"pagination": pagination,
|
|
**guide,
|
|
**(await _filter_options(db)),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/scopes/filter", response_class=HTMLResponse)
|
|
async def scopes_filter(
|
|
request: Request,
|
|
description: str = "",
|
|
manufacturer_id: str = "",
|
|
scope_model_id: str = "",
|
|
sub_part_type_id: str = "",
|
|
fits_part_type_id: str = "",
|
|
page: int = 1,
|
|
page_size: int = _PAGE_SIZE,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User | None = Depends(_optional_scopes_user),
|
|
):
|
|
filters = {
|
|
"description": description,
|
|
"manufacturer_id": manufacturer_id,
|
|
"scope_model_id": scope_model_id,
|
|
"sub_part_type_id": sub_part_type_id,
|
|
"fits_part_type_id": fits_part_type_id,
|
|
}
|
|
only_exact = user is None
|
|
guide, pagination = await _paginated_guide(
|
|
db, filters, page, page_size, only_exact=only_exact
|
|
)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"_scopes_table_card.html",
|
|
{
|
|
**guide,
|
|
"pagination": pagination,
|
|
"can_write_scopes": _can_write_scopes(user),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/scopes/bulk-delete", response_class=HTMLResponse)
|
|
async def bulk_delete_scopes(
|
|
request: Request,
|
|
scope_ids: list[int] = Form([]),
|
|
description: str = Form(""),
|
|
manufacturer_id: str = Form(""),
|
|
scope_model_id: str = Form(""),
|
|
sub_part_type_id: str = Form(""),
|
|
fits_part_type_id: str = Form(""),
|
|
db: AsyncSession = Depends(get_db),
|
|
actor: User = Depends(web_require("augmented_parts:write")),
|
|
):
|
|
for augmented_part_id in scope_ids:
|
|
ap = await augmented_part_crud.get(db, augmented_part_id)
|
|
if ap is None:
|
|
continue
|
|
label = ap.sku or ap.description
|
|
await augmented_part_crud.delete(db, ap)
|
|
await audit_crud.log(
|
|
db,
|
|
actor=actor,
|
|
action="delete",
|
|
resource_type="AugmentedPart",
|
|
resource_id=augmented_part_id,
|
|
description=audit_crud.describe_delete(
|
|
"AugmentedPart", augmented_part_id, label
|
|
),
|
|
)
|
|
filters = {
|
|
"description": description,
|
|
"manufacturer_id": manufacturer_id,
|
|
"scope_model_id": scope_model_id,
|
|
"sub_part_type_id": sub_part_type_id,
|
|
"fits_part_type_id": fits_part_type_id,
|
|
}
|
|
guide, pagination = await _paginated_guide(db, filters, 1, _PAGE_SIZE)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"_scopes_table_card.html",
|
|
{
|
|
**guide,
|
|
"pagination": pagination,
|
|
"can_write_scopes": _can_write_scopes(actor),
|
|
},
|
|
)
|