""" API — the web front door for the pricing engine ================================================ FastAPI turns Python functions into web endpoints. A web app (Phase 3) or a curl command sends a request; this file checks it, runs the engine, and sends back JSON. browser / curl ──> api.py ──> catalog_engine.py │ └────> db.py ──> Supabase (price_items, shops) Four endpoints: GET /health "are you alive, and can you reach the database?" GET /price-book the whole catalog, for populating a dropdown POST /quote send a job, get back a priced quote POST /jobs/{id}/price price a job SAVED IN SUPABASE and write the prices back (n8n calls this automatically) When the server is on the internet, set QUOTE_API_KEY in the environment and every endpoint except /health requires the header X-API-Key: . Leave it unset for local work and the door stays open. Run it: uvicorn api:app --reload Then open http://127.0.0.1:8000/docs — FastAPI builds a clickable test page for every endpoint automatically. Pydantic models (the classes below) are the inspection station: every request is checked against them BEFORE our code runs. Wrong type, missing field, fee out of range — the caller gets a clear 422 error and the engine never sees bad data. """ import os from typing import Annotated, Literal, Optional, Union from fastapi import Depends, FastAPI, Header, HTTPException from pydantic import BaseModel, Field, ValidationError from catalog_engine import index_price_book, build_quote from db import (fetch_price_items, fetch_shop_settings, fetch_job, fetch_job_lines, save_job_pricing, save_job_pricing_error) app = FastAPI( title="Quote Gen API", description="Pricing engine for custom cabinet estimates.", version="0.2.0", ) # --------------------------------------------------------------------- # REQUEST SHAPES — what a caller is allowed to send to POST /quote # `kind` tells Pydantic which of the two line shapes to check against # (a "discriminated union"). Literal["catalog"] means: this field must # be exactly the word "catalog". # --------------------------------------------------------------------- class CatalogLine(BaseModel): """A line from the price book. Price comes from the database.""" kind: Literal["catalog"] sku: str qty: float = Field(default=1, gt=0) # gt=0 -> must be positive with_glass: bool = False class BuyoutLine(BaseModel): """Something you buy and resell. You enter COST; the engine marks it up.""" kind: Literal["buyout"] description: str cost: float = Field(ge=0) # ge=0 -> zero or more qty: float = Field(default=1, gt=0) # "Look at `kind`, then check against CatalogLine or BuyoutLine accordingly." JobLine = Annotated[Union[CatalogLine, BuyoutLine], Field(discriminator="kind")] class QuoteRequest(BaseModel): """The whole job. Percent overrides are optional; shop defaults apply.""" customer: str project: str gc_referral: bool = False # Shop policy: finder fee is 8-10%. Anything else is rejected. gc_fee_pct: Optional[float] = Field(default=None, ge=8, le=10) # Buy-out markup is adjustable per job (shop default is 40%). buyout_markup_pct: Optional[float] = Field(default=None, ge=0) # Sales tax: apply_tax=false turns it off for a tax-exempt job; # tax_pct overrides the shop's rate (default 6%) when it is on. apply_tax: bool = True tax_pct: Optional[float] = Field(default=None, ge=0) lines: list[JobLine] = Field(min_length=1) # at least one line # --------------------------------------------------------------------- # API KEY — the lock on the front door (only when QUOTE_API_KEY is set) # FastAPI "dependencies" run before the endpoint. This one compares the # X-API-Key header with the key in the environment and rejects mismatches # with 401 before any pricing code runs. # --------------------------------------------------------------------- def require_api_key(x_api_key: Optional[str] = Header(default=None)): expected = os.environ.get("QUOTE_API_KEY", "") if expected and x_api_key != expected: raise HTTPException(status_code=401, detail="Missing or wrong X-API-Key") # --------------------------------------------------------------------- # ENDPOINTS # --------------------------------------------------------------------- @app.get("/health") def health(): """Liveness check. Also reports whether the database answers.""" try: settings = fetch_shop_settings() return {"status": "ok", "database": "ok", "shop": settings["shop_name"]} except Exception as e: # noqa: BLE001 - report, don't crash return {"status": "ok", "database": "unreachable", "detail": str(e)} @app.get("/price-book", dependencies=[Depends(require_api_key)]) def price_book(): """Every item in the price book, sorted by category then SKU.""" items = fetch_price_items() return {"count": len(items), "items": items} @app.post("/quote", dependencies=[Depends(require_api_key)]) def quote(req: QuoteRequest): """Price a job. Body must match QuoteRequest (see /docs for the shape).""" book = index_price_book(fetch_price_items()) settings = fetch_shop_settings() # model_dump() turns the validated Pydantic object back into the # plain dict the engine expects. The engine never sees Pydantic. try: return build_quote(req.model_dump(), book, settings) except LookupError as e: # Unknown SKU is the caller's mistake -> 400 Bad Request, with the SKU. raise HTTPException(status_code=400, detail=str(e)) @app.post("/jobs/{job_id}/price", dependencies=[Depends(require_api_key)]) def price_saved_job(job_id: str): """Price a job that lives in Supabase and write the numbers back. This is the automation entry point. A database trigger tells n8n "job X changed", n8n calls this, and the job's lines get their unit_price / line_total while the job gets subtotal, finder_fee, tax, total and priced_at. Re-running it is harmless: same inputs, same numbers (idempotent). """ job = fetch_job(job_id) if job is None: raise HTTPException(status_code=404, detail=f"Job {job_id} not found") rows = fetch_job_lines(job_id) # A job whose lines were all deleted is simply worth nothing. if not rows: empty = {"lines": [], "subtotal": 0.0, "finder_fee": 0.0, "tax": 0.0, "total": 0.0} save_job_pricing(job_id, [], empty) return {"job_id": job_id, **empty} # Reshape database rows into the same body POST /quote accepts, so the # same validation rules (fee 8-10%, cost required, ...) apply here. tax_pct = float(job.get("tax_pct") or 0) body = { "customer": job.get("customer") or "", "project": job.get("project") or "", "gc_referral": bool(job.get("gc_referral")), "gc_fee_pct": job.get("gc_fee_pct"), "buyout_markup_pct": job.get("buyout_markup_pct"), "apply_tax": tax_pct > 0, # tax_pct of 0 in the DB = tax-exempt "tax_pct": tax_pct if tax_pct > 0 else None, "lines": [], } for row in rows: if row["kind"] == "catalog": body["lines"].append({"kind": "catalog", "sku": row.get("sku"), "qty": row.get("qty") or 1, "with_glass": bool(row.get("with_glass"))}) else: body["lines"].append({"kind": "buyout", "description": row.get("description") or "", "cost": row.get("cost"), "qty": row.get("qty") or 1}) try: req = QuoteRequest.model_validate(body) quote = build_quote(req.model_dump(), index_price_book(fetch_price_items()), fetch_shop_settings()) except (ValidationError, LookupError) as e: # Bad input (unknown SKU, fee out of range, buy-out with no cost): # remember WHY on the job so the app can show it, then say 400. if isinstance(e, ValidationError): # "lines.2.buyout.cost: Input should be a valid number" beats # Pydantic's generic "1 validation error for QuoteRequest". message = "; ".join( ".".join(str(p) for p in err["loc"]) + ": " + err["msg"] for err in e.errors())[:500] else: message = str(e)[:500] save_job_pricing_error(job_id, message) raise HTTPException(status_code=400, detail=message) save_job_pricing(job_id, [row["id"] for row in rows], quote) return {"job_id": job_id, **quote}