PacketStation/server/packetstation/api/checklists.py

"""Per-department checklists (change C8, Gitea issue 23).

Every signed-in session reads them. A department's supervisor edits that department's list and Admin edits any;
a lead, a worker, Q.C. and the office roles do not. A save needs a password session, carries the version it started
from (`base`, or If-Match) and writes one audit row (`checklists.dept_updated`). A change reaches only stages that
have not started: a stage freezes its list at START. `items: null` sets the department back to the default for its
kind (Admin's weld or plain list at /api/admin/checklists). The Q.C. check items are not here.
"""

from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Depends, Request
from fastapi.encoders import jsonable_encoder
from psycopg.types.json import Jsonb
from pydantic import BaseModel, ConfigDict, Field

from .. import audit, policy
from ..domain import checklists, packets
from ..policy.custody import SUPERVISOR
from .deps import Db, Principal, current_principal, get_db, require_password_session
from .errors import ApiError
from .packets import WRITE

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

MSG_NOT_YOURS = "Only this department's supervisor or an Admin can change its checklist."
MSG_STALE = "Someone else changed this checklist. Reload and try again."
MSG_NO_DEPT = "No such shop department."


class DeptChecklistPut(BaseModel):
    model_config = ConfigDict(extra="forbid")

    items: list[Any] | None = Field(max_length=checklists.ITEMS_MAX * 2)
    base: int | None = Field(None, ge=0)


def can_edit(pp: policy.Principal, dept_code: str) -> bool:
    """Admin, or the department's Supervisor (a lead is not enough). Q.C. never edits a department list."""
    return pp.admin or pp.level(dept_code) >= SUPERVISOR


def _shop_departments(conn: Any) -> list[dict[str, Any]]:
    return packets.rows(
        conn,
        "SELECT code, name, weld FROM departments WHERE kind = 'shop' AND active ORDER BY sort, code",
    )


def _dept(conn: Any, code: str) -> dict[str, Any]:
    d = packets.one(conn, "SELECT code, name, weld FROM departments WHERE code = %s AND kind = 'shop'", (code,))
    if d is None:
        raise ApiError(404, "not_found", MSG_NO_DEPT)
    return d


def _summary(conn: Any, pp: policy.Principal, d: dict[str, Any]) -> dict[str, Any]:
    own, version = checklists.load_dept(conn, d["code"])
    return {
        "dept_code": d["code"],
        "dept_name": d["name"],
        "weld": bool(d["weld"]),
        "custom": own is not None,
        "version": version,
        "editable": can_edit(pp, d["code"]),
    }


def _one(conn: Any, pp: policy.Principal, d: dict[str, Any]) -> dict[str, Any]:
    out = _summary(conn, pp, d)
    out["items"] = checklists.live_list(conn, d["code"], bool(d["weld"]))
    out["defaults"] = checklists.default_for(conn, bool(d["weld"]))
    return out


@router.get("")
def list_checklists(
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> list[dict[str, Any]]:
    """Every active shop department: whether it has its own list, the version and whether the caller may edit."""
    conn = d.conn
    pp = packets.policy_principal_for(conn, p)
    return [_summary(conn, pp, x) for x in _shop_departments(conn)]


@router.get("/{dept}")
def get_checklist(
    dept: str,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """One department's live list (its own, else the default for its kind), unfilled: {drawing}, {rev} and {wps}
    stay as placeholders. `defaults` is the default for its kind."""
    conn = d.conn
    pp = packets.policy_principal_for(conn, p)
    return _one(conn, pp, _dept(conn, dept[:10]))


@router.put("/{dept}", dependencies=WRITE)
def put_checklist(
    dept: str,
    body: DeptChecklistPut,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    conn = d.conn
    pp = packets.policy_principal_for(conn, p)
    row = _dept(conn, dept[:10])
    code = str(row["code"])
    if not can_edit(pp, code):
        raise ApiError(403, "forbidden", MSG_NOT_YOURS)
    require_password_session(p)
    base = body.base
    if base is None:
        tag = (request.headers.get("if-match") or "").strip().strip('"').removeprefix("W/").strip('"')
        if not tag.isdigit():
            raise ApiError(428, "if_match_required", "Send the version you started from.")
        base = int(tag)
    value: dict[str, Any]
    if body.items is None:
        value = {"items": None}
    else:
        clean, errors = checklists.check_list("checklist", body.items)
        if errors or clean is None:
            raise ApiError(422, "bad_checklist", errors[0], extra={"errors": errors})
        value = {"items": clean}
    key = checklists.dept_key(code)
    with conn.transaction():
        cur = conn.execute("SELECT version FROM settings WHERE key = %s FOR UPDATE", (key,)).fetchone()
        current = int(cur[0]) if cur else 0
        if base != current:
            raise ApiError(409, "version_conflict", MSG_STALE, extra={"current": jsonable_encoder(_one(conn, pp, row))})
        before = checklists.load_dept(conn, code)[0]
        conn.execute(
            "INSERT INTO settings (key, value, version, updated_at, updated_by) VALUES (%s, %s, 1, now(), %s)"
            " ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, version = settings.version + 1,"
            " updated_at = now(), updated_by = EXCLUDED.updated_by",
            (key, Jsonb(value), p.user_id),
        )
        audit.append(
            conn,
            action="checklists.dept_updated",
            user_id=p.user_id,
            session_id=p.session_id,
            role_snapshot=p.role_snapshot,
            subject_type="settings",
            subject_id=key,
            detail={"dept": code, "before": before, "after": value["items"]},
            ip=p.ip,
        )
    return _one(conn, pp, row)