PacketStation/server/packetstation/api/departments.py

"""Departments and their stations, and the people of a shop department (read-only)."""

from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Depends

from ..domain import shop
from .deps import Db, Principal, current_principal, get_db
from .errors import ApiError

router = APIRouter(prefix="/api", tags=["departments"])


@router.get("/departments")
def departments(_: Principal = Depends(current_principal), d: Db = Depends(get_db)) -> list[dict[str, Any]]:  # noqa: B008
    conn = d.conn
    rows = conn.execute(
        "SELECT code, slug, name, kind::text, weld FROM departments WHERE active ORDER BY sort, code"
    ).fetchall()
    stations = conn.execute(
        "SELECT dept_code, id, name FROM stations WHERE active ORDER BY dept_code, sort, name"
    ).fetchall()
    by_dept: dict[str, list[dict[str, Any]]] = {}
    for code, sid, name in stations:
        by_dept.setdefault(code, []).append({"id": int(sid), "name": name})
    return [
        {"code": r[0], "slug": r[1], "name": r[2], "kind": r[3], "weld": bool(r[4]), "stations": by_dept.get(r[0], [])}
        for r in rows
    ]


@router.get("/departments/{code}/people")
def department_people(
    code: str,
    _: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> list[dict[str, Any]]:
    """The active users with a shop role in a shop department: the assign sheet's welder picker. 404 for an
    unknown or support department."""
    conn = d.conn
    found = conn.execute(
        "SELECT code FROM departments WHERE code = %s AND kind = 'shop' AND active", (code[:10],)
    ).fetchone()
    if found is None:
        raise ApiError(404, "not_found", "No such shop department.")
    return shop.people(conn, str(found[0]))