REST API design for AI services

{
  "type": "https://example.com/problems/order-not-found",
  "title": "Blocked order not found",
  "status": 404,
  "detail": "There is no blocked order 9999 in the sample data.",
  "instance": "/v1/blocked-orders/9999",
  "request_id": "bef0da40d5654c2882032fa75985e76b"
}

event: delta
data: {"text": "Order "}
id: 0

event: done
data: {"id": "exp_c056d1f1c04f", "sales_order": "4714", "explanation": "...", "next_step": "fix_master_data", ...}

"""Unit 6: the blocked-orders AI API, redesigned with REST rules.

It shows the design choices an AI service needs, each on a real endpoint:
  - resources and verbs:   /v1/blocked-orders, /v1/explanations, /v1/explanation-jobs
  - versioning:            the major version in the path (/v1)
  - pagination:            ?limit= and an opaque ?cursor= on the list
  - idempotency:           an Idempotency-Key header on POST /v1/explanations
  - one error format:      application/problem+json (RFC 9457) for every error
  - long-running jobs:     202 Accepted, a Location header and a job you poll
  - streaming:             server-sent events on POST /v1/explanations/stream
  - a contract:            the OpenAPI spec, exported to a file

How to run (from your course folder, with .venv turned on):
    python unit06/designed_api.py --demo                         # walk through every rule, no server
    python unit06/designed_api.py --export-openapi unit06/openapi.json
    python unit06/designed_api.py                                # start the server on port 8000
It needs ORCHESTRATE_API_KEY in .env (from "Building an AI API"). The model is a made-up sample model.
Stop the server with Ctrl+C.
"""
import argparse
import base64
import hashlib
import json
import os
import re
import secrets
import sys
import threading
import time
import uuid
from collections.abc import Iterable
from typing import Literal, Optional

from fastapi import Depends, FastAPI, Header, Query, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.security import APIKeyHeader
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel, Field
from starlette.exceptions import HTTPException as StarletteHTTPException

API_VERSION = "1.1.0"           # the contract's version; the path only carries the major version (v1)
PROBLEM_BASE = "https://example.com/problems/"   # replace with a page that documents your error types

BLOCKED_ORDERS = {   # made-up data in the shape of the course's running example
    "4711": {"customer": "Made-up Retail GmbH", "reason": "Credit limit exceeded", "net_value": 12500.00, "currency": "EUR"},
    "4712": {"customer": "Example Foods Ltd", "reason": "Missing export documents", "net_value": 8300.50, "currency": "GBP"},
    "4713": {"customer": "Sample Tools Inc", "reason": "Credit limit exceeded", "net_value": 21000.00, "currency": "USD"},
    "4714": {"customer": "Demo Garden AG", "reason": "Incomplete delivery address", "net_value": 640.00, "currency": "CHF"},
    "4715": {"customer": "Fictional Motors SA", "reason": "Credit limit exceeded", "net_value": 47200.00, "currency": "EUR"},
    "4716": {"customer": "Placeholder Pharma BV", "reason": "Missing export documents", "net_value": 3150.75, "currency": "EUR"},
    "4717": {"customer": "Invented Textiles Srl", "reason": "Incomplete delivery address", "net_value": 980.00, "currency": "EUR"},
}
NEXT_STEP_BY_REASON = {"Credit limit exceeded": "credit_review", "Missing export documents": "complete_documents",
                       "Incomplete delivery address": "fix_master_data"}
NextStep = Literal["credit_review", "complete_documents", "fix_master_data", "contact_customer"]
SalesOrder = Field(pattern=r"^[0-9]{1,10}$", description="SAP sales order number, digits only", examples=["4711"])


# ---------- the contract: request and response shapes ----------

class BlockedOrder(BaseModel):
    sales_order: str
    customer: str
    reason: str
    net_value: float
    currency: str


class BlockedOrderPage(BaseModel):
    """One page of a list. next_cursor is null on the last page."""
    items: list[BlockedOrder]
    next_cursor: Optional[str] = Field(description="Pass this as ?cursor= to get the next page; null when done")


class ExplanationRequest(BaseModel):
    sales_order: str = SalesOrder
    max_words: int = Field(default=60, ge=20, le=120, description="Longest explanation you want, in words")


class Explanation(BaseModel):
    id: str
    sales_order: str
    explanation: str
    next_step: NextStep
    model: str
    created_at: str


class JobRequest(BaseModel):
    sales_orders: list[str] = Field(min_length=1, max_length=50, description="1 to 50 order numbers")


class Job(BaseModel):
    id: str
    status: Literal["queued", "running", "succeeded", "failed"]
    total: int
    done: int
    results: list[Explanation] = []
    errors: list[dict] = []


class Problem(BaseModel):
    """RFC 9457 problem details. Clients branch on type and status, never on the detail text."""
    type: str
    title: str
    status: int
    detail: str
    instance: str
    request_id: str


PROBLEM_RESPONSES = {   # documented in the OpenAPI spec for every operation that can fail this way
    401: {"model": Problem, "content": {"application/problem+json": {}}, "description": "Missing or wrong API key"},
    404: {"model": Problem, "content": {"application/problem+json": {}}, "description": "No such resource"},
    422: {"model": Problem, "content": {"application/problem+json": {}},
          "description": "Request does not match the contract, or an Idempotency-Key was reused with another body"},
}


class ApiProblem(Exception):
    def __init__(self, status: int, slug: str, title: str, detail: str, headers: Optional[dict] = None):
        self.status, self.slug, self.title, self.detail, self.headers = status, slug, title, detail, headers or {}


# ---------- the sample model ----------

class SampleModel:
    """Made-up answers so everything runs without an account. Swap in the real model from 'Building an AI API'."""
    name = "sample-model"

    def explain(self, sales_order: str, order: dict, max_words: int) -> dict:
        text = (f"Order {sales_order} for {order['customer']} is on hold because of: {order['reason'].lower()}. "
                f"It is worth {order['net_value']:,.2f} {order['currency']}. The responsible team must review it "
                "before it can ship.")
        words = text.split()[:max_words]
        return {"explanation": " ".join(words), "next_step": NEXT_STEP_BY_REASON.get(order["reason"], "contact_customer")}


# ---------- the app ----------

def create_app(api_key: str, model=None, job_seconds_per_order: float = 0.3) -> FastAPI:
    model = model or SampleModel()
    app = FastAPI(title="Orchestrate blocked-order explainer", version=API_VERSION,
                  description="Explains blocked SAP sales orders. Course example with made-up data. "
                              "Every error is application/problem+json (RFC 9457).")
    explanations: dict[str, dict] = {}          # explanation id -> explanation
    idempotency: dict[str, tuple[str, dict]] = {}   # Idempotency-Key -> (request fingerprint, stored result)
    jobs: dict[str, dict] = {}
    lock = threading.Lock()

    # --- one error format for everything ---
    def problem(request: Request, status: int, slug: str, title: str, detail: str, headers=None) -> JSONResponse:
        body = {"type": PROBLEM_BASE + slug, "title": title, "status": status, "detail": detail,
                "instance": request.url.path, "request_id": getattr(request.state, "request_id", "")}
        return JSONResponse(body, status_code=status, headers=headers, media_type="application/problem+json")

    @app.exception_handler(ApiProblem)
    async def handle_api_problem(request: Request, error: ApiProblem):
        return problem(request, error.status, error.slug, error.title, error.detail, error.headers)

    @app.exception_handler(RequestValidationError)
    async def handle_validation(request: Request, error: RequestValidationError):
        first = error.errors()[0]
        where = ".".join(str(part) for part in first["loc"])
        return problem(request, 422, "invalid-request", "Request does not match the contract", f"{where}: {first['msg']}")

    @app.exception_handler(StarletteHTTPException)
    async def handle_http(request: Request, error: StarletteHTTPException):   # e.g. unknown paths
        return problem(request, error.status_code, "http-error", "HTTP error", str(error.detail))

    @app.middleware("http")
    async def request_id(request: Request, call_next):
        incoming = request.headers.get("X-Request-ID", "")
        request.state.request_id = incoming if re.fullmatch(r"[A-Za-z0-9-]{8,64}", incoming) else uuid.uuid4().hex
        response = await call_next(request)
        response.headers["X-Request-ID"] = request.state.request_id
        return response

    key_header = APIKeyHeader(name="X-API-Key", auto_error=False, description="Your ORCHESTRATE_API_KEY")

    def require_key(key: Optional[str] = Depends(key_header)) -> None:
        if not key or not secrets.compare_digest(key.encode(), api_key.encode()):
            raise ApiProblem(401, "unauthorized", "Missing or wrong API key", "Send your key in the X-API-Key header.",
                             {"WWW-Authenticate": "APIKey"})

    def find_order(sales_order: str) -> dict:
        order = BLOCKED_ORDERS.get(sales_order)
        if order is None:
            raise ApiProblem(404, "order-not-found", "Blocked order not found",
                             f"There is no blocked order {sales_order} in the sample data.")
        return order

    def make_explanation(sales_order: str, max_words: int) -> dict:
        answer = model.explain(sales_order, find_order(sales_order), max_words)
        return {"id": "exp_" + uuid.uuid4().hex[:12], "sales_order": sales_order, **answer, "model": model.name,
                "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}

    secured = [Depends(require_key)]

    @app.get("/health", tags=["service"], operation_id="getHealth", summary="Is the service up?")
    def health() -> dict:
        return {"status": "ok", "version": API_VERSION}

    # --- pagination: limit plus an opaque cursor ---
    @app.get("/v1/blocked-orders", response_model=BlockedOrderPage, dependencies=secured, tags=["blocked orders"],
             operation_id="listBlockedOrders", summary="List blocked orders, one page at a time",
             responses={401: PROBLEM_RESPONSES[401], 422: PROBLEM_RESPONSES[422]})
    def list_blocked_orders(limit: int = Query(default=3, ge=1, le=100, description="Items per page"),
                            cursor: Optional[str] = Query(default=None, description="next_cursor from the previous page"),
                            reason: Optional[str] = Query(default=None, description="Filter by block reason")):
        numbers = sorted(n for n, o in BLOCKED_ORDERS.items() if reason is None or o["reason"] == reason)
        after = ""
        if cursor:
            try:
                after = json.loads(base64.urlsafe_b64decode(cursor.encode()))["after"]
            except Exception:
                raise ApiProblem(422, "invalid-cursor", "Invalid cursor", "Use the next_cursor value exactly as returned.")
        remaining = [n for n in numbers if n > after]
        page = remaining[:limit]
        next_cursor = None
        if len(remaining) > limit:   # the cursor hides the position, so you can change how it works later
            next_cursor = base64.urlsafe_b64encode(json.dumps({"after": page[-1]}).encode()).decode()
        return {"items": [{"sales_order": n, **BLOCKED_ORDERS[n]} for n in page], "next_cursor": next_cursor}

    @app.get("/v1/blocked-orders/{sales_order}", response_model=BlockedOrder, dependencies=secured,
             tags=["blocked orders"], operation_id="getBlockedOrder", summary="Read one blocked order",
             responses={401: PROBLEM_RESPONSES[401], 404: PROBLEM_RESPONSES[404], 422: PROBLEM_RESPONSES[422]})
    def get_blocked_order(sales_order: str):
        return {"sales_order": sales_order, **find_order(sales_order)}

    # --- create a resource, safely retryable with Idempotency-Key ---
    @app.post("/v1/explanations", response_model=Explanation, status_code=201, dependencies=secured,
              tags=["explanations"], operation_id="createExplanation", summary="Explain one blocked order",
              responses={**PROBLEM_RESPONSES, 409: {"model": Problem, "content": {"application/problem+json": {}},
                                                     "description": "A request with this Idempotency-Key is still running"}})
    def create_explanation(body: ExplanationRequest, request: Request, response: Response,
                           idempotency_key: Optional[str] = Header(default=None, alias="Idempotency-Key",
                                                                   description="A new random value (UUID) per action. "
                                                                   "Retries with the same key and body return the first result.",
                                                                   pattern=r"^[A-Za-z0-9-]{8,64}$")):
        fingerprint = hashlib.sha256(body.model_dump_json().encode()).hexdigest()
        if idempotency_key:
            with lock:   # check and reserve the key in one step, so two parallel retries can't both run
                stored = idempotency.get(idempotency_key)
                if stored is None:
                    idempotency[idempotency_key] = (fingerprint, None)   # None: still in progress
            if stored:
                if stored[0] != fingerprint:
                    raise ApiProblem(422, "idempotency-key-reused", "Idempotency key already used",
                                     "This Idempotency-Key was used with a different request body. Use a new key.")
                if stored[1] is None:
                    raise ApiProblem(409, "request-in-progress", "Request still in progress",
                                     "A request with this Idempotency-Key is still running. Retry shortly.",
                                     {"Retry-After": "1"})
                response.headers["Location"] = f"/v1/explanations/{stored[1]['id']}"
                response.headers["Idempotent-Replay"] = "true"   # our own header: this is the stored result
                return stored[1]
        try:
            result = make_explanation(body.sales_order, body.max_words)
        except ApiProblem:
            if idempotency_key:
                with lock:
                    idempotency.pop(idempotency_key, None)   # a failed attempt may be retried with the same key
            raise
        with lock:
            explanations[result["id"]] = result
            if idempotency_key:
                idempotency[idempotency_key] = (fingerprint, result)
        response.headers["Location"] = f"/v1/explanations/{result['id']}"
        return result

    @app.get("/v1/explanations/{explanation_id}", response_model=Explanation, dependencies=secured,
             tags=["explanations"], operation_id="getExplanation", summary="Read an explanation again",
             responses={401: PROBLEM_RESPONSES[401], 404: PROBLEM_RESPONSES[404], 422: PROBLEM_RESPONSES[422]})
    def get_explanation(explanation_id: str):
        if explanation_id not in explanations:
            raise ApiProblem(404, "explanation-not-found", "Explanation not found", f"No explanation {explanation_id}.")
        return explanations[explanation_id]

    # --- streaming: server-sent events, then one final validated object ---
    def prepare_explanation(body: ExplanationRequest) -> dict:
        """Runs before the stream starts, so a missing order is still a normal 404 problem."""
        result = make_explanation(body.sales_order, body.max_words)
        explanations[result["id"]] = result
        return result

    @app.post("/v1/explanations/stream", response_class=EventSourceResponse, dependencies=secured,
              tags=["explanations"], operation_id="streamExplanation",
              summary="Explain one order, sending words as they are ready",
              responses={401: PROBLEM_RESPONSES[401], 404: PROBLEM_RESPONSES[404], 422: PROBLEM_RESPONSES[422]})
    def stream_explanation(result: dict = Depends(prepare_explanation)) -> Iterable[ServerSentEvent]:
        for number, word in enumerate(result["explanation"].split()):
            time.sleep(0.05)   # pretend the model writes word by word
            yield ServerSentEvent(event="delta", id=str(number), data={"text": word + " "})
        yield ServerSentEvent(event="done", data=result)   # the full, schema-checked object comes last

    # --- long-running work: 202 Accepted, Location, poll the job ---
    def run_job(job_id: str) -> None:
        job = jobs[job_id]
        job["status"] = "running"
        for number in job["sales_orders"]:
            time.sleep(job_seconds_per_order)   # pretend each explanation takes a while
            try:
                job["results"].append(make_explanation(number, 60))
            except ApiProblem as error:
                job["errors"].append({"sales_order": number, "type": PROBLEM_BASE + error.slug, "detail": error.detail})
            job["done"] += 1
        job["status"] = "succeeded" if job["results"] else "failed"

    @app.post("/v1/explanation-jobs", response_model=Job, status_code=202, dependencies=secured,
              tags=["jobs"], operation_id="createExplanationJob", summary="Explain many orders in the background",
              responses={401: PROBLEM_RESPONSES[401], 422: PROBLEM_RESPONSES[422]})
    def create_job(body: JobRequest, response: Response):
        job_id = "job_" + uuid.uuid4().hex[:12]
        jobs[job_id] = {"id": job_id, "status": "queued", "total": len(body.sales_orders), "done": 0,
                        "results": [], "errors": [], "sales_orders": body.sales_orders}
        accepted = dict(jobs[job_id])   # what the caller sees now: queued, nothing done yet
        threading.Thread(target=run_job, args=(job_id,), daemon=True).start()
        response.headers["Location"] = f"/v1/explanation-jobs/{job_id}"
        response.headers["Retry-After"] = "1"   # seconds the caller should wait before the first poll
        return accepted

    @app.get("/v1/explanation-jobs/{job_id}", response_model=Job, dependencies=secured, tags=["jobs"],
             operation_id="getExplanationJob", summary="Check a background job",
             responses={401: PROBLEM_RESPONSES[401], 404: PROBLEM_RESPONSES[404], 422: PROBLEM_RESPONSES[422]})
    def get_job(job_id: str, response: Response):
        if job_id not in jobs:
            raise ApiProblem(404, "job-not-found", "Job not found", f"No job {job_id}.")
        if jobs[job_id]["status"] in ("queued", "running"):
            response.headers["Retry-After"] = "1"
        return jobs[job_id]

    def openapi_with_problems() -> dict:
        """FastAPI files error schemas under the success media type; move them to application/problem+json."""
        if app.openapi_schema is None:
            from fastapi.openapi.utils import get_openapi
            spec = get_openapi(title=app.title, version=app.version, description=app.description, routes=app.routes)
            for path in spec["paths"].values():
                for operation in path.values():
                    for code, answer in operation.get("responses", {}).items():
                        if "application/problem+json" in answer.get("content", {}):
                            answer["content"] = {"application/problem+json": {
                                "schema": {"$ref": "#/components/schemas/Problem"}}}
            app.openapi_schema = spec
        return app.openapi_schema

    app.openapi = openapi_with_problems
    return app


# ---------- demo: every rule, in memory, no server ----------

def demo(api_key: str) -> None:
    from fastapi.testclient import TestClient
    client = TestClient(create_app(api_key, job_seconds_per_order=0.2))
    key = {"X-API-Key": api_key}

    def show(title: str, response, fields=None) -> None:
        body = response.json() if "json" in response.headers.get("content-type", "") else response.text
        if fields and isinstance(body, dict):
            body = {f: body.get(f) for f in fields}
        print(f"\n== {title}\n{response.status_code} {response.headers.get('content-type', '')}")
        for header in ("Location", "Retry-After", "Idempotent-Replay"):
            if header in response.headers:
                print(f"{header}: {response.headers[header]}")
        print(json.dumps(body, indent=2)[:600])

    page1 = client.get("/v1/blocked-orders?limit=3", headers=key)
    show("1. First page of blocked orders", page1, ["next_cursor"])
    print("orders:", [o["sales_order"] for o in page1.json()["items"]])
    page2 = client.get("/v1/blocked-orders", params={"limit": 3, "cursor": page1.json()["next_cursor"]}, headers=key)
    print("next page orders:", [o["sales_order"] for o in page2.json()["items"]])

    show("2. An error, in one format for every error (RFC 9457)", client.get("/v1/blocked-orders/9999", headers=key))

    idem = {**key, "Idempotency-Key": str(uuid.uuid4())}
    first = client.post("/v1/explanations", json={"sales_order": "4711"}, headers=idem)
    show("3. Create an explanation (201 Created)", first, ["id", "next_step", "model"])
    again = client.post("/v1/explanations", json={"sales_order": "4711"}, headers=idem)
    show("4. Same request retried with the same Idempotency-Key", again, ["id"])
    print("same id as before:", again.json()["id"] == first.json()["id"])
    show("5. Same key, different body", client.post("/v1/explanations", json={"sales_order": "4712"}, headers=idem),
         ["status", "title"])

    created = client.post("/v1/explanation-jobs", json={"sales_orders": ["4711", "4712", "9999"]}, headers=key)
    show("6. Start a background job (202 Accepted)", created, ["id", "status", "total", "done"])
    location = created.headers["Location"]
    while True:
        time.sleep(float(created.headers.get("Retry-After", "1")) / 2)
        status = client.get(location, headers=key).json()
        print(f"   poll: {status['status']} {status['done']}/{status['total']}")
        if status["status"] in ("succeeded", "failed"):
            break
    print(f"   results: {len(status['results'])}, errors: {[e['sales_order'] for e in status['errors']]}")

    print("\n== 7. Stream an explanation (server-sent events)")
    with client.stream("POST", "/v1/explanations/stream", json={"sales_order": "4714"},
                       headers=key) as stream:
        lines = [line for line in stream.iter_lines() if line.startswith(("event:", "data:"))]
    print("\n".join(lines[:4]) + "\n...\n" + "\n".join(lines[-2:])[:300])

    show("8. No key", client.get("/v1/blocked-orders"), ["status", "title"])
    print("\nDemo finished. Next: export the contract with --export-openapi unit06/openapi.json")


def main() -> None:
    parser = argparse.ArgumentParser(description="The Unit 6 blocked-orders API, designed with REST rules.")
    parser.add_argument("--demo", action="store_true", help="walk through every rule in memory, no server")
    parser.add_argument("--export-openapi", metavar="FILE", help="write the OpenAPI spec to FILE and stop")
    parser.add_argument("--port", type=int, default=8000, help="port to listen on (default 8000)")
    args = parser.parse_args()

    from dotenv import load_dotenv
    load_dotenv()
    api_key = os.environ.get("ORCHESTRATE_API_KEY", "")
    if args.export_openapi:
        spec = create_app(api_key or "not-needed-for-export").openapi()
        with open(args.export_openapi, "w", encoding="utf-8") as file:
            json.dump(spec, file, indent=2)
        print(f"Wrote {args.export_openapi}: OpenAPI {spec['openapi']}, {len(spec['paths'])} paths, "
              f"API version {spec['info']['version']}")
        return
    if len(api_key) < 20:
        sys.exit("ORCHESTRATE_API_KEY is missing or shorter than 20 characters in .env. "
                 "See 'Building an AI API', Step 2.")
    if args.demo:
        demo(api_key)
        return
    import uvicorn
    print(f"Open http://127.0.0.1:{args.port}/docs in your browser. Press Ctrl+C to stop.")
    uvicorn.run(create_app(api_key), host="127.0.0.1", port=args.port)


if __name__ == "__main__":
    main()

== 1. First page of blocked orders
200 application/json
{
  "next_cursor": "eyJhZnRlciI6ICI0NzEzIn0="
}
orders: ['4711', '4712', '4713']
next page orders: ['4714', '4715', '4716']

== 2. An error, in one format for every error (RFC 9457)
404 application/problem+json
{
  "type": "https://example.com/problems/order-not-found",
  "title": "Blocked order not found",
  "status": 404,
  "detail": "There is no blocked order 9999 in the sample data.",
  "instance": "/v1/blocked-orders/9999",
  "request_id": "bef0da40d5654c2882032fa75985e76b"
}

== 3. Create an explanation (201 Created)
201 application/json
Location: /v1/explanations/exp_72f0c532e21c
...
== 4. Same request retried with the same Idempotency-Key
201 application/json
Location: /v1/explanations/exp_72f0c532e21c
Idempotent-Replay: true
...
same id as before: True

== 5. Same key, different body
422 application/problem+json
...
== 6. Start a background job (202 Accepted)
202 application/json
Location: /v1/explanation-jobs/job_31e3163166d3
Retry-After: 1
...
   poll: running 2/3
   poll: succeeded 3/3
   results: 2, errors: ['9999']

== 7. Stream an explanation (server-sent events)
event: delta
data: {"text": "Order "}
...
event: done
data: {"id": "exp_c056d1f1c04f", "sales_order": "4714", ...}

== 8. No key
401 application/problem+json
...
Demo finished. Next: export the contract with --export-openapi unit06/openapi.json

Wrote unit06/openapi.json: OpenAPI 3.1.0, 8 paths, API version 1.1.0

"""Unit 6: check an OpenAPI spec against the course's API design rules.

Built-in modules only. It reads the JSON spec that designed_api.py exports and prints one line per rule:
OK when every operation follows it, FAIL with the operations that don't.

How to run (from your course folder, with .venv turned on):
    python unit06/lint_api.py unit06/openapi.json
"""
import json
import re
import sys

METHODS = {"get", "put", "post", "patch", "delete"}
UNVERSIONED = {"/health"}   # paths that may live outside /v1


def operations(spec: dict):
    for path, item in spec.get("paths", {}).items():
        for method, operation in item.items():
            if method in METHODS:
                yield path, method, operation


def rule_versioned_paths(spec):
    """Every business path starts with a major version, such as /v1/."""
    return [p for p in spec["paths"] if p not in UNVERSIONED and not re.match(r"^/v[0-9]+/", p)]


def rule_resource_names(spec):
    """Path segments are lowercase nouns with hyphens; no verbs such as /getOrders."""
    bad = []
    for path in spec["paths"]:
        for segment in path.strip("/").split("/"):
            if segment.startswith("{"):
                continue
            if not re.fullmatch(r"[a-z0-9]+(-[a-z0-9]+)*", segment) or re.match(r"^(get|create|update|delete|do)", segment):
                bad.append(f"{path} ({segment})")
    return bad


def rule_operation_ids(spec):
    """Every operation has a unique operationId and a summary, so generated clients get good names."""
    seen, bad = set(), []
    for path, method, op in operations(spec):
        op_id = op.get("operationId")
        if not op_id or op_id in seen or not op.get("summary"):
            bad.append(f"{method.upper()} {path}")
        seen.add(op_id)
    return bad


def rule_problem_errors(spec):
    """Every documented 4xx or 5xx response uses application/problem+json."""
    bad = []
    for path, method, op in operations(spec):
        for code, response in op.get("responses", {}).items():
            if code[0] in "45" and list(response.get("content", {})) != ["application/problem+json"]:
                bad.append(f"{method.upper()} {path} {code}")
    return bad


def rule_secured_operations_document_401(spec):
    """If the API has a security scheme, every secured operation documents 401."""
    bad = []
    for path, method, op in operations(spec):
        if op.get("security") and "401" not in op.get("responses", {}):
            bad.append(f"{method.upper()} {path}")
    return bad


def rule_lists_are_paginated(spec):
    """Every GET on a collection (no {id} at the end) takes limit and cursor."""
    bad = []
    for path, method, op in operations(spec):
        if method == "get" and path not in UNVERSIONED and not path.endswith("}"):
            names = {p.get("name") for p in op.get("parameters", [])}
            if not {"limit", "cursor"} <= names:
                bad.append(f"GET {path}")
    return bad


def rule_creates_are_retry_safe(spec):
    """Every POST that answers 201 Created accepts an Idempotency-Key header."""
    bad = []
    for path, method, op in operations(spec):
        if method == "post" and "201" in op.get("responses", {}):
            names = {p.get("name") for p in op.get("parameters", []) if p.get("in") == "header"}
            if "Idempotency-Key" not in names:
                bad.append(f"POST {path}")
    return bad


RULES = [rule_versioned_paths, rule_resource_names, rule_operation_ids, rule_problem_errors,
         rule_secured_operations_document_401, rule_lists_are_paginated, rule_creates_are_retry_safe]


def main() -> None:
    if len(sys.argv) != 2:
        sys.exit("Usage: python unit06/lint_api.py unit06/openapi.json")
    try:
        with open(sys.argv[1], encoding="utf-8") as file:
            spec = json.load(file)
    except FileNotFoundError:
        sys.exit(f"{sys.argv[1]} not found. Export it first: python unit06/designed_api.py --export-openapi {sys.argv[1]}")
    print(f"Checking {spec['info']['title']} {spec['info']['version']} (OpenAPI {spec['openapi']}), "
          f"{len(list(operations(spec)))} operations")
    failed = 0
    for rule in RULES:
        problems = rule(spec)
        label = rule.__doc__.strip().split("\n")[0]
        print(f"{'OK  ' if not problems else 'FAIL'}  {label}")
        for item in problems:
            print(f"        {item}")
        failed += bool(problems)
    print("All rules pass." if not failed else f"{failed} rule(s) failed.")
    sys.exit(1 if failed else 0)


if __name__ == "__main__":
    main()

Checking Orchestrate blocked-order explainer 1.1.0 (OpenAPI 3.1.0), 8 operations
OK    Every business path starts with a major version, such as /v1/.
OK    Path segments are lowercase nouns with hyphens; no verbs such as /getOrders.
OK    Every operation has a unique operationId and a summary, so generated clients get good names.
OK    Every documented 4xx or 5xx response uses application/problem+json.
OK    If the API has a security scheme, every secured operation documents 401.
OK    Every GET on a collection (no {id} at the end) takes limit and cursor.
OK    Every POST that answers 201 Created accepts an Idempotency-Key header.
All rules pass.

FAIL  Every documented 4xx or 5xx response uses application/problem+json.
        GET /v1/blocked-orders/{sales_order} 422

Open http://127.0.0.1:8000/docs in your browser. Press Ctrl+C to stop.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

event: delta
data: {"text": "Order "}
id: 0

event: delta
data: {"text": "4712 "}
id: 1
...
event: done
data: {"id": "exp_...", "sales_order": "4712", ...}

HTTP/1.1 202 Accepted
location: /v1/explanation-jobs/job_62379b22b817
retry-after: 1
...
{"id":"job_62379b22b817","status":"queued","total":2,"done":0,"results":[],"errors":[]}