Validation and error handling

import logging
import re
from typing import Literal

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator
from starlette.exceptions import HTTPException as StarletteHTTPException

logger = logging.getLogger("brewbean")

MENU = {
    "espresso": 120,
    "cappuccino": 180,
    "cold-brew": 220,
    "filter-coffee": 80,
    "masala-chai": 90,
    "croissant": 150,
    "banana-bread": 130,
    "veg-sandwich": 160,
}

app = FastAPI(title="Brew & Bean validation demo")
orders: dict[int, dict] = {}


# ---------- Models: the rules live here ----------
class OrderItem(BaseModel):
    item_id: str
    qty: int = Field(ge=1, le=20)

    @field_validator("item_id")
    @classmethod
    def item_must_exist(cls, value: str) -> str:
        if value not in MENU:
            raise ValueError(f"unknown menu item '{value}'")
        return value


class OrderCreate(BaseModel):
    customer_name: str = Field(min_length=2, max_length=60)
    phone: str
    items: list[OrderItem] = Field(min_length=1, max_length=20)
    pickup_time: str | None = Field(default=None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$")

    @field_validator("customer_name")
    @classmethod
    def clean_name(cls, value: str) -> str:
        return value.strip()

    @field_validator("phone")
    @classmethod
    def indian_mobile(cls, value: str) -> str:
        digits = re.sub(r"[\s-]", "", value)
        if not re.fullmatch(r"[6-9]\d{9}", digits):
            raise ValueError("must be a 10-digit Indian mobile number")
        return digits


class StatusUpdate(BaseModel):
    status: Literal["preparing", "ready", "picked_up", "cancelled"]


# ---------- One error shape for every failure ----------
class AppError(Exception):
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message


def error_body(code: str, message: str, details: list | None = None) -> dict:
    body = {"error": {"code": code, "message": message}}
    if details:
        body["error"]["details"] = details
    return body


@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(status_code=exc.status_code, content=error_body(exc.code, exc.message))


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    details = [
        {
            "field": ".".join(str(part) for part in err["loc"][1:]),
            "message": err["msg"],
        }
        for err in exc.errors()
    ]
    return JSONResponse(
        status_code=422,
        content=error_body("validation_error", "Some fields are invalid", details),
    )


@app.exception_handler(StarletteHTTPException)
async def http_error_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(status_code=exc.status_code, content=error_body("http_error", str(exc.detail)))


@app.exception_handler(Exception)
async def unexpected_error_handler(request: Request, exc: Exception):
    logger.exception("Unhandled error on %s %s", request.method, request.url.path)
    # The client gets no internals; the log has the full traceback.
    return JSONResponse(status_code=500, content=error_body("internal_error", "Something went wrong"))


# ---------- Routes ----------
@app.post("/orders", status_code=201)
def create_order(body: OrderCreate):
    total = sum(MENU[i.item_id] * i.qty for i in body.items)  # the server computes the total
    order_id = len(orders) + 1
    orders[order_id] = {
        "id": order_id,
        "customer_name": body.customer_name,
        "phone": body.phone,
        "items": [i.model_dump() for i in body.items],
        "total": total,
        "status": "received",
        "pickup_time": body.pickup_time,
    }
    return orders[order_id]


@app.patch("/orders/{order_id}/status")
def update_status(order_id: int, body: StatusUpdate):
    order = orders.get(order_id)
    if order is None:
        raise AppError(404, "order_not_found", f"Order {order_id} does not exist")
    if order["status"] in ("picked_up", "cancelled"):
        raise AppError(409, "order_closed", f"Order is already {order['status']}")
    order["status"] = body.status
    return order


@app.get("/crash")
def crash():
    return 1 / 0  # simulates a bug, to see the 500 handler

# A valid order
curl -X POST http://127.0.0.1:8000/orders -H "Content-Type: application/json" -d '{"customer_name":"Asha","phone":"98765 43210","items":[{"item_id":"cappuccino","qty":2}],"pickup_time":"09:30"}'

# An invalid order
curl -X POST http://127.0.0.1:8000/orders -H "Content-Type: application/json" -d '{"customer_name":"A","phone":"12345","items":[{"item_id":"pizza","qty":0}],"pickup_time":"25:00"}'

# An order that does not exist
curl -X PATCH http://127.0.0.1:8000/orders/99/status -H "Content-Type: application/json" -d '{"status":"ready"}'

# A bug
curl http://127.0.0.1:8000/crash

// npm install express zod
import express from "express";
import { z } from "zod";

const OrderItem = z.object({
  item_id: z.string(),
  qty: z.number().int().min(1).max(20),
});
const OrderCreate = z.object({
  customer_name: z.string().trim().min(2).max(60),
  items: z.array(OrderItem).min(1).max(20),
});

const app = express();
app.use(express.json());

app.post("/orders", (req, res) => {
  const result = OrderCreate.safeParse(req.body);
  if (!result.success) {
    return res.status(400).json({
      error: { code: "validation_error", message: "Some fields are invalid", details: result.error.issues },
    });
  }
  res.status(201).json(result.data);
});

// Error-handling middleware: four arguments, registered last
app.use((err, req, res, next) => {
  console.error(err);
  res.status(500).json({ error: { code: "internal_error", message: "Something went wrong" } });
});

app.listen(3000);

I am building a [describe endpoint, e.g. POST /bookings for a salon app] with these fields: [list fields and what they mean]. Write Pydantic v2 models (using Field constraints and field_validator) that validate it strictly. Then list 8 test inputs, both valid and invalid, with the exact error each should produce.