"""Projects: list, read, create, field edits, renumber, soft delete; and the number lookup. Reads need any signed-in session (PM included). Writes need Clerical (clerical or clerical_supervisor); renumber and delete need clerical_supervisor. An admin outside those roles may do any of them with an `it_reason` of at least 8 characters, kept on the audit row. The router has no prefix because it also serves GET /api/lookup/number; every path starts with /api. """ from __future__ import annotations from typing import Any from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, ConfigDict, Field from ..domain import projects as dom from ..text import reason_ok from .deps import Db, Principal, current_principal, get_db, require_password_session from .errors import ApiError router = APIRouter(tags=["projects"]) PREFIX = "/api/projects" CLERICAL = frozenset({"clerical", "clerical_supervisor"}) CLERICAL_SUP = frozenset({"clerical_supervisor"}) # Generous input caps; clean_text cuts every value to its field limit. TEXT = 4000 def authorize(p: Principal, *, supervisor: bool, it_reason: str | None) -> dom.Who: """The write check. Returns who is writing; it_reason is set only for an admin acting outside their roles.""" need = CLERICAL_SUP if supervisor else CLERICAL if need & set(p.roles): return dom.Who(user_id=p.user_id, session_id=p.session_id, role_snapshot=p.role_snapshot, ip=p.ip) if p.is_admin: require_password_session(p) why = reason_ok(it_reason) if why is None: raise ApiError(400, "it_reason_required", "Give an IT reason of at least 8 characters.") return dom.Who( user_id=p.user_id, session_id=p.session_id, role_snapshot=p.role_snapshot, ip=p.ip, it_reason=why ) if supervisor: raise ApiError(403, "forbidden", "Only a Clerical Supervisor can do this.") raise ApiError(403, "forbidden", "Only Clerical can change projects and units.") class ProjectCreate(BaseModel): number: str = Field(max_length=200) name: str | None = Field(None, max_length=TEXT) customer: str | None = Field(None, max_length=TEXT) po: str | None = Field(None, max_length=TEXT) due: str | None = Field(None, max_length=40) status: str | None = Field(None, max_length=20) notes: str | None = Field(None, max_length=TEXT) it_reason: str | None = Field(None, max_length=TEXT) class FieldCas(BaseModel): """`{set, base}`: the new values and what the client last saw for them.""" model_config = ConfigDict(populate_by_name=True) values: dict[str, Any] = Field(alias="set") base: dict[str, Any] = Field(default_factory=dict) it_reason: str | None = Field(None, max_length=TEXT) class RenumberProject(BaseModel): number: str = Field(max_length=200) reason: str | None = Field(None, max_length=TEXT) it_reason: str | None = Field(None, max_length=TEXT) class ReasonBody(BaseModel): reason: str | None = Field(None, max_length=TEXT) it_reason: str | None = Field(None, max_length=TEXT) class UnitsCreate(BaseModel): lot: str | None = Field(None, max_length=40) lots: list[str] | None = Field(None, max_length=dom.MAX_UNITS_AT_ONCE) count: int | None = None start: str | None = Field(None, max_length=40) name: str | None = Field(None, max_length=TEXT) due: str | None = Field(None, max_length=40) status: str | None = Field(None, max_length=20) notes: str | None = Field(None, max_length=TEXT) it_reason: str | None = Field(None, max_length=TEXT) @router.get(PREFIX) def list_projects( p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 q: str = Query("", max_length=200), status: str = Query("active", max_length=20), ) -> list[dict[str, Any]]: return dom.list_projects(d.conn, q=q, status=status) @router.post(PREFIX, status_code=201) def create_project( body: ProjectCreate, p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: who = authorize(p, supervisor=False, it_reason=body.it_reason) return dom.create_project(d.conn, who, body.model_dump()) @router.get(PREFIX + "/{project_id}") def project_detail( project_id: int, p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: return dom.project_detail(d.conn, project_id) @router.patch(PREFIX + "/{project_id}") def update_project( project_id: int, body: FieldCas, p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: who = authorize(p, supervisor=False, it_reason=body.it_reason) return dom.update_project(d.conn, who, project_id, body.values, body.base) @router.post(PREFIX + "/{project_id}/renumber") def renumber_project( project_id: int, body: RenumberProject, p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: who = authorize(p, supervisor=True, it_reason=body.it_reason) return dom.renumber_project(d.conn, who, project_id, body.number, body.reason) @router.delete(PREFIX + "/{project_id}") def remove_project( project_id: int, body: ReasonBody, p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: who = authorize(p, supervisor=True, it_reason=body.it_reason) return dom.remove_project(d.conn, who, project_id, body.reason) @router.post(PREFIX + "/{project_id}/units", status_code=201) def create_units( project_id: int, body: UnitsCreate, p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> list[dict[str, Any]]: who = authorize(p, supervisor=False, it_reason=body.it_reason) return dom.create_units(d.conn, who, project_id, body.model_dump()) @router.get("/api/lookup/number") def lookup_number( p: Principal = Depends(current_principal), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 n: str = Query(..., min_length=1, max_length=40), ) -> dict[str, Any]: found = dom.lookup(d.conn, n) if found is None: raise ApiError(404, "not_found", "No project or unit has that number.") return found