PacketStation/server/packetstation/api/shop.py

"""The shop screens: the supervisor board, the stations a person may work at, the stall board of one station, and
the Stall screen's own list (/api/stall/mine: the packets assigned to me, across stations).

Any signed-in session reads them; the board answers 403 for someone who leads no department. They read
PostgreSQL only.
"""

from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Depends, Query, Request

from ..domain import packets, search, shop
from ..text import clean_text
from .deps import Db, Principal, current_principal, get_db
from .packets import FIND, site_today

router = APIRouter(prefix="/api/shop", tags=["shop"])
stall_router = APIRouter(prefix="/api/stall", tags=["shop"])


@router.get("/board")
def board(
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
    find: dict[str, Any] = FIND,
    project_id: int | None = Query(None),
    unit_id: int | None = Query(None),
) -> dict[str, Any]:
    """The supervisor board. `dept` picks one of the departments the person leads (it scopes the board and its
    KPIs); the other find bar values narrow the cards."""
    conn = d.conn
    pp = packets.policy_principal_for(conn, p)
    dept = clean_text(find.get("dept") or "", 10) or None
    filters = search.parse_filters(
        {**find, "dept": None, "project_id": project_id, "unit_id": unit_id},
        pp,
        screen="supervisor",
        today=site_today(request),
    )
    return shop.board(conn, pp, dept=dept, project_id=project_id, unit_id=unit_id, filters=filters)


@router.get("/stations")
def stations(
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> list[dict[str, Any]]:
    conn = d.conn
    return shop.stations_for(conn, packets.policy_principal_for(conn, p))


@router.get("/stall")
def stall(
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
    station_id: int = Query(ge=1),
    find: dict[str, Any] = FIND,
) -> dict[str, Any]:
    """The cards at one station. The find bar values narrow them; `station_id` here is the stall's station, so the
    Station filter does not apply. Mine is the station's own work."""
    conn = d.conn
    pp = packets.policy_principal_for(conn, p)
    raw = {**find, "station_id": None, "mine_station_id": find.get("mine_station_id") or station_id}
    filters = search.parse_filters(raw, pp, screen="stall", today=site_today(request))
    return shop.stall(conn, pp, station_id, filters)


@stall_router.get("/mine")
def stall_mine(
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
    find: dict[str, Any] = FIND,
) -> dict[str, Any]:
    """The Stall screen (change C11): the packets assigned to the signed-in person, each at the station the
    supervisor picked, across stations. No station in the request; the session decides. The find bar values narrow
    the cards (Mine is these cards already)."""
    conn = d.conn
    pp = packets.policy_principal_for(conn, p)
    filters = search.parse_filters(find, pp, screen="stall", today=site_today(request))
    return shop.stall_mine(conn, pp, filters)