"""Procurement (采购) — suppliers, purchase/restock requests and receiving. Mirrors the factory's paper flow (耗材補貨清單 / 包材補貨清單): raise a restock request, assign a person-in-charge, move it through a status, receive the goods. Receiving is the point where this module meets inventory: it never keeps its own stock count. POST /api/procurement/requests/{id}/receive writes a real `inventory_transactions` row (reason='purchase') and bumps `inventory_items.current_stock` in the same transaction, exactly the way app/routes/inventory.py's POST /api/inventory/transactions does — inventory stays the single source of truth for on-hand quantities. """ import json from datetime import date from decimal import Decimal, InvalidOperation from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from app.auth import require_permission from app.audit import record_after, record_before, snapshot as audit_snapshot from app.database import get_connection from app.errors import raise_app_error router = APIRouter() REQUEST_STATUSES = ("draft", "requested", "approved", "ordered", "received", "cancelled") REQUEST_CATEGORIES = ("consumable", "packaging", "equipment", "other") OPEN_STATUSES = ("draft", "requested", "approved", "ordered") ON_ORDER_STATUSES = ("approved", "ordered") # Arbitrary but stable key for pg_advisory_xact_lock, used to serialise request_no # generation. Transaction-scoped: released automatically on commit/rollback. REQUEST_NO_LOCK_KEY = 5251001 # --------------------------------------------------------------------- models -- class SupplierCreateRequest(BaseModel): name: str contact: Optional[str] = None phone: Optional[str] = None email: Optional[str] = None address: Optional[str] = None lead_time_days: Optional[int] = None note: Optional[str] = None class SupplierUpdateRequest(BaseModel): name: Optional[str] = None contact: Optional[str] = None phone: Optional[str] = None email: Optional[str] = None address: Optional[str] = None lead_time_days: Optional[int] = None note: Optional[str] = None is_active: Optional[bool] = None class PurchaseRequestItemInput(BaseModel): inventory_item_id: Optional[int] = None item_name: str spec: Optional[str] = None qty: float = 0 unit: str = "pcs" unit_cost: Optional[float] = None note: Optional[str] = None class PurchaseRequestItemUpdate(BaseModel): inventory_item_id: Optional[int] = None item_name: Optional[str] = None spec: Optional[str] = None qty: Optional[float] = None unit: Optional[str] = None unit_cost: Optional[float] = None note: Optional[str] = None class PurchaseRequestCreate(BaseModel): title: str category: str = "consumable" supplier_id: Optional[int] = None requested_by: Optional[str] = None pic: Optional[str] = None status: str = "draft" needed_by: Optional[date] = None total_cost: Optional[float] = None currency: str = "CNY" note: Optional[str] = None items: List[PurchaseRequestItemInput] = [] class PurchaseRequestUpdate(BaseModel): title: Optional[str] = None category: Optional[str] = None supplier_id: Optional[int] = None requested_by: Optional[str] = None pic: Optional[str] = None status: Optional[str] = None needed_by: Optional[date] = None total_cost: Optional[float] = None currency: Optional[str] = None note: Optional[str] = None class ReceiveLine(BaseModel): item_id: int qty_received: float class ReceiveRequest(BaseModel): lines: List[ReceiveLine] # Optional but strongly recommended from the UI: the UNIQUE constraint on # platform_purchase_receipts.idempotency_key is what turns a double-submitted # receive into a no-op instead of double-counting stock. idempotency_key: Optional[str] = None note: Optional[str] = None # -------------------------------------------------------------------- helpers -- def _dec(value) -> Decimal: """Decimal from anything the DB or pydantic hands us, without float drift.""" if value is None: return Decimal("0") if isinstance(value, Decimal): return value try: return Decimal(str(value)) except (InvalidOperation, ValueError): return Decimal("0") def _generate_request_no(cur) -> str: """Collision-safe `PR-YYYYMMDD-NNN`. Deliberately NOT a row count. A transaction-scoped advisory lock serialises concurrent creators, so two simultaneous POSTs cannot read the same max and produce the same number; the UNIQUE constraint on request_no backstops it. """ cur.execute("SELECT pg_advisory_xact_lock(%s)", (REQUEST_NO_LOCK_KEY,)) prefix = "PR-{}-".format(date.today().strftime("%Y%m%d")) cur.execute( """ SELECT request_no FROM platform_purchase_requests WHERE request_no LIKE %s ORDER BY request_no DESC LIMIT 1 """, (prefix + "%",), ) row = cur.fetchone() seq = 1 if row: try: seq = int(str(row["request_no"]).rsplit("-", 1)[1]) + 1 except (ValueError, IndexError): seq = 1 return "{}{:03d}".format(prefix, seq) def _fetch_request(cur, request_id: int) -> dict: cur.execute( """ SELECT r.*, s.name AS supplier_name, s.contact AS supplier_contact, s.phone AS supplier_phone, s.email AS supplier_email, s.lead_time_days AS supplier_lead_time_days FROM platform_purchase_requests r LEFT JOIN platform_suppliers s ON s.id = r.supplier_id WHERE r.id = %s """, (request_id,), ) row = cur.fetchone() if not row: raise_app_error("ERR-PROC-001", detail={"request_id": request_id}) return dict(row) def _fetch_items(cur, request_id: int) -> list: cur.execute( """ SELECT pi.*, ii.name AS inventory_item_name, ii.current_stock AS inventory_current_stock FROM platform_purchase_request_items pi LEFT JOIN inventory_items ii ON ii.id = pi.inventory_item_id WHERE pi.request_id = %s ORDER BY pi.id ASC """, (request_id,), ) return [dict(r) for r in cur.fetchall()] def _request_detail(cur, request_id: int) -> dict: request = _fetch_request(cur, request_id) items = _fetch_items(cur, request_id) total_qty = sum(_dec(i["qty"]) for i in items) received_qty = sum(_dec(i["received_qty"]) for i in items) items_cost = sum(_dec(i["qty"]) * _dec(i["unit_cost"]) for i in items) request["items"] = items request["item_count"] = len(items) request["total_qty"] = total_qty request["received_qty_total"] = received_qty request["items_cost"] = items_cost return request def _touch_request(cur, request_id: int) -> None: cur.execute( "UPDATE platform_purchase_requests SET updated_at = now() WHERE id = %s", (request_id,), ) # ------------------------------------------------------------------ suppliers -- @router.get("/api/procurement/suppliers") def list_suppliers( is_active: Optional[bool] = None, q: Optional[str] = None, current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: sql = """ SELECT s.*, (SELECT COUNT(*) FROM platform_purchase_requests r WHERE r.supplier_id = s.id) AS request_count FROM platform_suppliers s WHERE 1 = 1 """ params = [] if is_active is not None: sql += " AND s.is_active = %s" params.append(is_active) if q: sql += " AND (s.name ILIKE %s OR s.contact ILIKE %s OR s.phone ILIKE %s)" params.extend(["%" + q + "%"] * 3) sql += " ORDER BY s.is_active DESC, s.name ASC" cur.execute(sql, params) return cur.fetchall() except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.post("/api/procurement/suppliers") def create_supplier( req: SupplierCreateRequest, current_user: dict = Depends(require_permission("procurement.manage")), ): if not req.name.strip(): raise_app_error("ERR-PROC-009", detail={"field": "name"}, message_override="name is required") conn = get_connection() try: with conn.cursor() as cur: cur.execute("SELECT id FROM platform_suppliers WHERE name = %s", (req.name.strip(),)) if cur.fetchone(): raise_app_error("ERR-PROC-005", detail={"name": req.name}) cur.execute( """ INSERT INTO platform_suppliers (name, contact, phone, email, address, lead_time_days, note) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING * """, ( req.name.strip(), req.contact, req.phone, req.email, req.address, req.lead_time_days, req.note, ), ) supplier = cur.fetchone() conn.commit() return supplier except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.patch("/api/procurement/suppliers/{supplier_id}") def update_supplier( supplier_id: int, req: SupplierUpdateRequest, current_user: dict = Depends(require_permission("procurement.manage")), ): updates = req.model_dump(exclude_unset=True) if not updates: raise_app_error("ERR-PROC-010") conn = get_connection() try: with conn.cursor() as cur: if updates.get("name"): cur.execute( "SELECT id FROM platform_suppliers WHERE name = %s AND id <> %s", (updates["name"], supplier_id), ) if cur.fetchone(): raise_app_error("ERR-PROC-005", detail={"name": updates["name"]}) audit_snapshot(conn, "platform_suppliers", supplier_id, entity_type="procurement/suppliers", label_column="name") set_clause = ", ".join("{} = %s".format(key) for key in updates) values = list(updates.values()) + [supplier_id] cur.execute( "UPDATE platform_suppliers SET {}, updated_at = now() WHERE id = %s RETURNING *".format(set_clause), values, ) supplier = cur.fetchone() if not supplier: raise_app_error("ERR-PROC-002", detail={"supplier_id": supplier_id}) record_after("procurement/suppliers", supplier_id, supplier, label=supplier.get("name")) conn.commit() return supplier except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.delete("/api/procurement/suppliers/{supplier_id}") def delete_supplier( supplier_id: int, current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: cur.execute("SELECT id FROM platform_suppliers WHERE id = %s", (supplier_id,)) if not cur.fetchone(): raise_app_error("ERR-PROC-002", detail={"supplier_id": supplier_id}) cur.execute( "SELECT COUNT(*) AS n FROM platform_purchase_requests WHERE supplier_id = %s", (supplier_id,), ) if cur.fetchone()["n"]: # Deleting would orphan the purchasing history. Deactivating keeps the # supplier out of pickers while leaving old requests readable. raise_app_error("ERR-PROC-011", detail={"supplier_id": supplier_id}) audit_snapshot(conn, "platform_suppliers", supplier_id, entity_type="procurement/suppliers", label_column="name", action="delete") cur.execute("DELETE FROM platform_suppliers WHERE id = %s", (supplier_id,)) conn.commit() return {"deleted": True, "id": supplier_id} except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() # ----------------------------------------------------------- purchase requests -- @router.get("/api/procurement/requests") def list_purchase_requests( status: Optional[str] = None, category: Optional[str] = None, supplier_id: Optional[int] = None, pic: Optional[str] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, q: Optional[str] = None, current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: sql = """ SELECT r.*, s.name AS supplier_name, COALESCE(agg.item_count, 0) AS item_count, COALESCE(agg.total_qty, 0) AS total_qty, COALESCE(agg.received_qty_total, 0) AS received_qty_total, COALESCE(agg.items_cost, 0) AS items_cost, (r.needed_by IS NOT NULL AND r.needed_by < CURRENT_DATE AND r.status NOT IN ('received', 'cancelled')) AS is_overdue FROM platform_purchase_requests r LEFT JOIN platform_suppliers s ON s.id = r.supplier_id LEFT JOIN ( SELECT request_id, COUNT(*) AS item_count, SUM(qty) AS total_qty, SUM(received_qty) AS received_qty_total, SUM(qty * COALESCE(unit_cost, 0)) AS items_cost FROM platform_purchase_request_items GROUP BY request_id ) agg ON agg.request_id = r.id WHERE 1 = 1 """ params = [] if status: sql += " AND r.status = %s" params.append(status) if category: sql += " AND r.category = %s" params.append(category) if supplier_id: sql += " AND r.supplier_id = %s" params.append(supplier_id) if pic: sql += " AND r.pic ILIKE %s" params.append("%" + pic + "%") # Date window is on created_at (when the request was raised). if start_date: sql += " AND r.created_at >= %s" params.append(start_date) if end_date: sql += " AND r.created_at < (%s::date + 1)" params.append(end_date) if q: sql += """ AND ( r.request_no ILIKE %s OR r.title ILIKE %s OR r.note ILIKE %s OR r.requested_by ILIKE %s OR r.pic ILIKE %s OR EXISTS ( SELECT 1 FROM platform_purchase_request_items pi WHERE pi.request_id = r.id AND pi.item_name ILIKE %s ) )""" params.extend(["%" + q + "%"] * 6) sql += " ORDER BY r.created_at DESC, r.id DESC" cur.execute(sql, params) return cur.fetchall() except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.get("/api/procurement/requests/{request_id}") def get_purchase_request( request_id: int, current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: return _request_detail(cur, request_id) except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.post("/api/procurement/requests") def create_purchase_request( req: PurchaseRequestCreate, current_user: dict = Depends(require_permission("procurement.manage")), ): if not req.title.strip(): raise_app_error("ERR-PROC-009", detail={"field": "title"}, message_override="title is required") if req.category not in REQUEST_CATEGORIES: raise_app_error("ERR-PROC-006", detail={"allowed": sorted(REQUEST_CATEGORIES)}) if req.status not in REQUEST_STATUSES: raise_app_error("ERR-PROC-007", detail={"allowed": sorted(REQUEST_STATUSES)}) conn = get_connection() try: with conn.cursor() as cur: if req.supplier_id is not None: cur.execute("SELECT id FROM platform_suppliers WHERE id = %s", (req.supplier_id,)) if not cur.fetchone(): raise_app_error("ERR-PROC-002") request_no = _generate_request_no(cur) cur.execute( """ INSERT INTO platform_purchase_requests (request_no, title, category, supplier_id, requested_by, pic, status, needed_by, ordered_at, received_at, total_cost, currency, note) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, CASE WHEN %s IN ('ordered', 'received') THEN now() END, CASE WHEN %s = 'received' THEN now() END, %s, %s, %s) RETURNING * """, ( request_no, req.title.strip(), req.category, req.supplier_id, req.requested_by, req.pic, req.status, req.needed_by, req.status, req.status, req.total_cost, req.currency, req.note, ), ) request_row = cur.fetchone() new_id = request_row["id"] for item in req.items: if not item.item_name.strip(): raise_app_error("ERR-PROC-009", detail={"field": "item_name"}, message_override="Each item needs an item_name") if item.qty < 0: raise_app_error("ERR-PROC-008", message_override="Item qty cannot be negative") if item.inventory_item_id is not None: cur.execute("SELECT id FROM inventory_items WHERE id = %s", (item.inventory_item_id,)) if not cur.fetchone(): raise_app_error( "ERR-PROC-004", detail={"inventory_item_id": item.inventory_item_id}, ) cur.execute( """ INSERT INTO platform_purchase_request_items (request_id, inventory_item_id, item_name, spec, qty, unit, unit_cost, note) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, ( new_id, item.inventory_item_id, item.item_name.strip(), item.spec, item.qty, item.unit or "pcs", item.unit_cost, item.note, ), ) detail = _request_detail(cur, new_id) conn.commit() return detail except HTTPException: conn.rollback() raise except Exception as e: conn.rollback() raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.patch("/api/procurement/requests/{request_id}") def update_purchase_request( request_id: int, req: PurchaseRequestUpdate, current_user: dict = Depends(require_permission("procurement.manage")), ): updates = req.model_dump(exclude_unset=True) if not updates: raise_app_error("ERR-PROC-010") if "category" in updates and updates["category"] not in REQUEST_CATEGORIES: raise_app_error("ERR-PROC-006", detail={"allowed": sorted(REQUEST_CATEGORIES)}) if "status" in updates and updates["status"] not in REQUEST_STATUSES: raise_app_error("ERR-PROC-007", detail={"allowed": sorted(REQUEST_STATUSES)}) conn = get_connection() try: with conn.cursor() as cur: cur.execute( "SELECT * FROM platform_purchase_requests WHERE id = %s FOR UPDATE", (request_id,), ) existing = cur.fetchone() if not existing: raise_app_error("ERR-PROC-001") # Row already fetched FOR UPDATE -- reuse it rather than re-SELECTing. record_before("procurement/requests", request_id, existing, label=existing.get("title")) if updates.get("supplier_id") is not None: cur.execute("SELECT id FROM platform_suppliers WHERE id = %s", (updates["supplier_id"],)) if not cur.fetchone(): raise_app_error("ERR-PROC-002") set_parts = ["{} = %s".format(key) for key in updates] values = list(updates.values()) # Status timestamps are stamped automatically, but never clobber one that # is already set — a request that was ordered, then edited, keeps its # original ordered_at. new_status = updates.get("status") if new_status in ("ordered", "received") and existing["ordered_at"] is None: set_parts.append("ordered_at = now()") if new_status == "received" and existing["received_at"] is None: set_parts.append("received_at = now()") set_parts.append("updated_at = now()") cur.execute( "UPDATE platform_purchase_requests SET {} WHERE id = %s".format(", ".join(set_parts)), values + [request_id], ) cur.execute("SELECT * FROM platform_purchase_requests WHERE id = %s", (request_id,)) record_after("procurement/requests", request_id, cur.fetchone()) detail = _request_detail(cur, request_id) conn.commit() return detail except HTTPException: conn.rollback() raise except Exception as e: conn.rollback() raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.delete("/api/procurement/requests/{request_id}") def delete_purchase_request( request_id: int, current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: # Items and receipts cascade. Any inventory_transactions already written by # a receive are deliberately left alone: they are an immutable audit trail of # stock that physically moved, and deleting the paperwork must not silently # unwind the stock count. audit_snapshot(conn, "platform_purchase_requests", request_id, entity_type="procurement/requests", label_column="title", action="delete") cur.execute("DELETE FROM platform_purchase_requests WHERE id = %s RETURNING id", (request_id,)) if not cur.fetchone(): raise_app_error("ERR-PROC-001") conn.commit() return {"deleted": True, "id": request_id} except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() # ----------------------------------------------------------------- line items -- @router.post("/api/procurement/requests/{request_id}/items") def add_purchase_request_item( request_id: int, req: PurchaseRequestItemInput, current_user: dict = Depends(require_permission("procurement.manage")), ): if not req.item_name.strip(): raise_app_error("ERR-PROC-009", detail={"field": "item_name"}, message_override="item_name is required") if req.qty < 0: raise_app_error("ERR-PROC-008", detail={"qty": req.qty}, message_override="qty cannot be negative") conn = get_connection() try: with conn.cursor() as cur: cur.execute("SELECT id FROM platform_purchase_requests WHERE id = %s", (request_id,)) if not cur.fetchone(): raise_app_error("ERR-PROC-001") if req.inventory_item_id is not None: cur.execute("SELECT id FROM inventory_items WHERE id = %s", (req.inventory_item_id,)) if not cur.fetchone(): raise_app_error("ERR-PROC-004") cur.execute( """ INSERT INTO platform_purchase_request_items (request_id, inventory_item_id, item_name, spec, qty, unit, unit_cost, note) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING * """, ( request_id, req.inventory_item_id, req.item_name.strip(), req.spec, req.qty, req.unit or "pcs", req.unit_cost, req.note, ), ) item = cur.fetchone() _touch_request(cur, request_id) conn.commit() return item except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.patch("/api/procurement/items/{item_id}") def update_purchase_request_item( item_id: int, req: PurchaseRequestItemUpdate, current_user: dict = Depends(require_permission("procurement.manage")), ): updates = req.model_dump(exclude_unset=True) if not updates: raise_app_error("ERR-PROC-010") conn = get_connection() try: with conn.cursor() as cur: cur.execute( "SELECT * FROM platform_purchase_request_items WHERE id = %s FOR UPDATE", (item_id,), ) existing = cur.fetchone() if not existing: raise_app_error("ERR-PROC-003", detail={"item_id": item_id}) record_before("procurement/items", item_id, existing, label=existing.get("item_name")) if "qty" in updates and updates["qty"] is not None: if updates["qty"] < 0: raise_app_error("ERR-PROC-008", message_override="qty cannot be negative") # received_qty is only ever moved by the receive endpoint; refuse an edit # that would leave a line recorded as over-received. if _dec(updates["qty"]) < _dec(existing["received_qty"]): raise_app_error( "ERR-PROC-015", detail={ "qty": updates["qty"], "received_qty": existing["received_qty"], }, ) if updates.get("inventory_item_id") is not None: cur.execute("SELECT id FROM inventory_items WHERE id = %s", (updates["inventory_item_id"],)) if not cur.fetchone(): raise_app_error("ERR-PROC-004") set_clause = ", ".join("{} = %s".format(key) for key in updates) cur.execute( "UPDATE platform_purchase_request_items SET {} WHERE id = %s RETURNING *".format(set_clause), list(updates.values()) + [item_id], ) item = cur.fetchone() record_after("procurement/items", item_id, item, label=(item or {}).get("item_name")) _touch_request(cur, existing["request_id"]) conn.commit() return item except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() @router.delete("/api/procurement/items/{item_id}") def delete_purchase_request_item( item_id: int, current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: audit_snapshot(conn, "platform_purchase_request_items", item_id, entity_type="procurement/items", label_column="item_name", action="delete") cur.execute( "DELETE FROM platform_purchase_request_items WHERE id = %s RETURNING request_id", (item_id,), ) row = cur.fetchone() if not row: raise_app_error("ERR-PROC-003", detail={"item_id": item_id}) _touch_request(cur, row["request_id"]) conn.commit() return {"deleted": True, "id": item_id} except HTTPException: raise except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() # -------------------------------------------------------------------- receive -- @router.post("/api/procurement/requests/{request_id}/receive") def receive_purchase_request( request_id: int, req: ReceiveRequest, current_user: dict = Depends(require_permission("procurement.manage")), ): """Record goods arriving against a request, in ONE transaction. For each line: bump platform_purchase_request_items.received_qty and, when the line is linked to an inventory item, write an inventory_transactions row (reason='purchase') and bump inventory_items.current_stock — the same pair of writes app/routes/inventory.py performs, so stock and its audit log stay in step. Idempotency: pass `idempotency_key`. The UNIQUE constraint on platform_purchase_receipts.idempotency_key makes a replayed submit return the current state untouched instead of double-counting. Without a key, over-receiving past the ordered qty is still rejected, but two identical legitimate submits are indistinguishable from two real partial deliveries. """ if not req.lines: raise_app_error("ERR-PROC-009", detail={"field": "lines"}, message_override="lines cannot be empty") conn = get_connection() try: with conn.cursor() as cur: if req.idempotency_key: cur.execute( "SELECT id, request_id FROM platform_purchase_receipts WHERE idempotency_key = %s", (req.idempotency_key,), ) replay = cur.fetchone() if replay: detail = _request_detail(cur, replay["request_id"]) detail["replayed"] = True return detail cur.execute( "SELECT * FROM platform_purchase_requests WHERE id = %s FOR UPDATE", (request_id,), ) request_row = cur.fetchone() if not request_row: raise_app_error("ERR-PROC-001") if request_row["status"] == "cancelled": raise_app_error("ERR-PROC-013", detail={"request_id": request_id}) # Lock every line of the request (not just the ones being received) so a # concurrent receive on the same request serialises behind this one and the # "is everything received?" check below can't race. cur.execute( "SELECT * FROM platform_purchase_request_items WHERE request_id = %s ORDER BY id FOR UPDATE", (request_id,), ) items = {row["id"]: dict(row) for row in cur.fetchall()} # Collapse duplicate line ids in one payload before validating, so # {id:1, 5} + {id:1, 5} against a qty of 8 is rejected rather than half-applied. wanted = {} for line in req.lines: if line.item_id not in items: raise_app_error( "ERR-PROC-014", detail={"item_id": line.item_id, "request_id": request_id}, ) qty = _dec(line.qty_received) if qty <= 0: raise_app_error("ERR-PROC-008", message_override="qty_received must be greater than zero") wanted[line.item_id] = wanted.get(line.item_id, Decimal("0")) + qty for item_id, qty in wanted.items(): item = items[item_id] if _dec(item["received_qty"]) + qty > _dec(item["qty"]): raise_app_error( "ERR-PROC-012", detail={ "item_id": item_id, "item_name": item["item_name"], "receiving": qty, "ordered": item["qty"], "already_received": item["received_qty"], }, ) # Always logged, key or not — the receipt row is the audit trail of who # received what; the key (when supplied) is what makes it a replay guard. cur.execute( """ INSERT INTO platform_purchase_receipts (request_id, idempotency_key, received_by, lines, note) VALUES (%s, %s, %s, %s::jsonb, %s) """, ( request_id, req.idempotency_key, current_user.get("id"), json.dumps({str(k): str(v) for k, v in wanted.items()}), req.note, ), ) transactions = [] for item_id, qty in wanted.items(): item = items[item_id] cur.execute( """ UPDATE platform_purchase_request_items SET received_qty = received_qty + %s WHERE id = %s RETURNING * """, (qty, item_id), ) items[item_id] = dict(cur.fetchone()) if item["inventory_item_id"] is None: continue note = "Procurement receive {} — {}".format(request_row["request_no"], item["item_name"]) if req.note: note += " ({})".format(req.note) cur.execute( """ INSERT INTO inventory_transactions (item_id, change_qty, reason, notes, created_by) VALUES (%s, %s, 'purchase', %s, %s) RETURNING * """, (item["inventory_item_id"], qty, note, current_user.get("id")), ) transactions.append(dict(cur.fetchone())) cur.execute( "UPDATE inventory_items SET current_stock = current_stock + %s WHERE id = %s", (qty, item["inventory_item_id"]), ) fully_received = all( _dec(i["received_qty"]) >= _dec(i["qty"]) for i in items.values() ) and bool(items) if fully_received and request_row["status"] != "received": cur.execute( """ UPDATE platform_purchase_requests SET status = 'received', received_at = COALESCE(received_at, now()), ordered_at = COALESCE(ordered_at, now()), updated_at = now() WHERE id = %s """, (request_id,), ) else: _touch_request(cur, request_id) detail = _request_detail(cur, request_id) detail["inventory_transactions"] = transactions detail["replayed"] = False conn.commit() return detail except HTTPException: conn.rollback() raise except Exception as e: conn.rollback() raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close() # -------------------------------------------------------------------- summary -- @router.get("/api/procurement/summary") def procurement_summary( current_user: dict = Depends(require_permission("procurement.manage")), ): conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ SELECT status, COUNT(*) AS count FROM platform_purchase_requests GROUP BY status """ ) by_status = {row["status"]: row["count"] for row in cur.fetchall()} # Value still on order: prefer the header total_cost the buyer entered, # fall back to the sum of the line costs when it hasn't been filled in. cur.execute( """ SELECT COALESCE(SUM( COALESCE(r.total_cost, ( SELECT COALESCE(SUM(pi.qty * COALESCE(pi.unit_cost, 0)), 0) FROM platform_purchase_request_items pi WHERE pi.request_id = r.id )) ), 0) AS value_on_order FROM platform_purchase_requests r WHERE r.status = ANY(%s) """, (list(ON_ORDER_STATUSES),), ) value_on_order = cur.fetchone()["value_on_order"] cur.execute( """ SELECT r.id, r.request_no, r.title, r.status, r.pic, r.needed_by, s.name AS supplier_name, (CURRENT_DATE - r.needed_by) AS days_overdue FROM platform_purchase_requests r LEFT JOIN platform_suppliers s ON s.id = r.supplier_id WHERE r.needed_by IS NOT NULL AND r.needed_by < CURRENT_DATE AND r.status NOT IN ('received', 'cancelled') ORDER BY r.needed_by ASC """ ) overdue = cur.fetchall() return { "by_status": {status: by_status.get(status, 0) for status in REQUEST_STATUSES}, "open_count": sum(by_status.get(status, 0) for status in OPEN_STATUSES), "value_on_order": value_on_order, "overdue_count": len(overdue), "overdue": overdue, } except Exception as e: raise_app_error("ERR-PROC-500", detail={"cause": str(e)}) finally: conn.close()