Python HTTP Frameworks

"""Issue quotes through FastAPI and the Chio ASGI middleware."""
import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict, Field, StrictStr
from chio_asgi import ChioASGIConfig, ChioASGIMiddleware
from pricing_store import QuoteStore


class QuoteRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    sku: StrictStr
    percent: float = Field(ge=0, le=100)


store = QuoteStore("fastapi")
app = FastAPI(title="Larkfield pricing", docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(
    ChioASGIMiddleware,
    config=ChioASGIConfig(
        sidecar_url=os.environ.get("CHIO_SIDECAR_URL", "http://127.0.0.1:9098"),
        max_body_bytes=8192,
        exclude_paths=frozenset({"/healthz"}),
        exclude_methods=frozenset(),
    ),
)


@app.get("/healthz")
def health():
    return {"status": "ready", "framework": "fastapi"}


@app.get("/v1/quotes")
def list_quotes():
    return {"quotes": store.list()}


@app.post("/v1/quote", status_code=201)
def quote(body: QuoteRequest):
    # The middleware adds the admission receipt ID as the X-Chio-Receipt header.
    try:
        return store.issue(body.sku, body.percent)
    except ValueError as error:
        raise HTTPException(400, str(error)) from error

"""Larkfield's price table and the quotes the service issues, shared by both frameworks."""
from decimal import ROUND_HALF_UP, Decimal
import sqlite3
from pathlib import Path

PRICES = {"sku-118": Decimal("24.00"), "sku-204": Decimal("58.00"), "sku-301": Decimal("140.00")}
CENT = Decimal("0.01")


class QuoteStore:
    def __init__(self, name):
        directory = Path(".pricing")
        directory.mkdir(mode=0o700, exist_ok=True)
        self.path = directory / f"{name}.db"
        with sqlite3.connect(self.path) as db:
            db.execute("CREATE TABLE IF NOT EXISTS quotes(id INTEGER PRIMARY KEY, sku TEXT NOT NULL, percent REAL NOT NULL, price TEXT NOT NULL)")

    def list(self):
        with sqlite3.connect(self.path) as db:
            rows = db.execute("SELECT id, sku, percent, price FROM quotes ORDER BY id LIMIT 100").fetchall()
        return [{"id": r[0], "sku": r[1], "percent": r[2], "price": r[3], "currency": "USD"} for r in rows]

    def issue(self, sku, percent):
        if sku not in PRICES:
            raise ValueError(f"unknown sku {sku}")
        if isinstance(percent, bool) or not isinstance(percent, (int, float)) or not 0 <= percent <= 100:
            raise ValueError("percent must be between 0 and 100")
        price = str((PRICES[sku] * (1 - Decimal(str(percent)) / 100)).quantize(CENT, ROUND_HALF_UP))
        with sqlite3.connect(self.path) as db:
            result = db.execute("INSERT INTO quotes(sku, percent, price) VALUES (?, ?, ?)", (sku, percent, price))
        return {"id": result.lastrowid, "sku": sku, "percent": percent, "price": price, "currency": "USD"}

"""The same pricing contract as a complete Django application."""
import json
import os
from django.conf import settings

settings.configure(
    DEBUG=False,
    SECRET_KEY="local-pricing-example-no-sessions",
    ALLOWED_HOSTS=["127.0.0.1", "localhost"],
    ROOT_URLCONF=__name__,
    MIDDLEWARE=["chio_django.ChioDjangoMiddleware"],
    CHIO_SIDECAR_URL=os.environ.get("CHIO_SIDECAR_URL", "http://127.0.0.1:9098"),
    CHIO_EXCLUDE_PATHS=["/healthz"],
    CHIO_EXCLUDE_METHODS=[],
    DATA_UPLOAD_MAX_MEMORY_SIZE=8192,
    USE_TZ=True,
)

import django
django.setup()
from django.core.wsgi import get_wsgi_application
from django.http import JsonResponse
from django.urls import path
from pricing_store import QuoteStore

store = QuoteStore("django")


def health(request):
    return JsonResponse({"status": "ready", "framework": "django"})


def quotes(request):
    return JsonResponse({"quotes": store.list()})


def quote(request):
    if request.method != "POST":
        return JsonResponse({"error": "method_not_allowed"}, status=405)
    try:
        payload = json.loads(request.body)
        if not isinstance(payload, dict) or set(payload) != {"sku", "percent"}:
            raise ValueError("body must contain exactly sku and percent")
        issued = store.issue(payload["sku"], payload["percent"])
    except (ValueError, UnicodeError) as error:
        return JsonResponse({"error": str(error)}, status=400)
    issued["receipt_id"] = request.chio_receipt["id"]
    return JsonResponse(issued, status=201)


urlpatterns = [path("healthz", health), path("v1/quotes", quotes), path("v1/quote", quote)]
application = get_wsgi_application()

if __name__ == "__main__":
    from wsgiref.simple_server import make_server
    with make_server("127.0.0.1", 8092, application) as server:
        print("Django pricing: http://127.0.0.1:8092", flush=True)
        server.serve_forever()