OpenAPI Sidecar

"""Larkfield's pricing API over HTTP, with SQLite persistence and no Chio dependency."""
import argparse
from decimal import ROUND_HALF_UP, Decimal
import json
import sqlite3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit

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


def row_to_quote(row):
    return {"id": row[0], "sku": row[1], "percent": row[2], "price": row[3], "currency": "USD"}


class Pricing(BaseHTTPRequestHandler):
    def setup(self):
        super().setup()
        self.connection.settimeout(10)

    def respond(self, status, payload):
        body = json.dumps(payload, separators=(",", ":")).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = urlsplit(self.path).path
        with sqlite3.connect(self.server.database) as database:
            if path == "/healthz":
                return self.respond(200, {"status": "ready"})
            if path == "/v1/quotes":
                rows = database.execute("SELECT id, sku, percent, price FROM quotes ORDER BY id LIMIT 100").fetchall()
                return self.respond(200, {"quotes": [row_to_quote(row) for row in rows]})
            if path.startswith(PREFIX) and path[len(PREFIX):].isdigit():
                row = database.execute("SELECT id, sku, percent, price FROM quotes WHERE id = ?", (int(path[len(PREFIX):]),)).fetchone()
                if row:
                    return self.respond(200, row_to_quote(row))
        self.respond(404, {"error": "quote_not_found"})

    def do_POST(self):
        if urlsplit(self.path).path != "/v1/quote":
            return self.respond(404, {"error": "route_not_found"})
        if self.headers.get("Transfer-Encoding"):
            return self.respond(400, {"error": "content_length_required"})
        try:
            length = int(self.headers.get("Content-Length", "0"))
            if not 0 < length <= 8192:
                return self.respond(413, {"error": "body_must_be_1_to_8192_bytes"})
            raw = self.rfile.read(length)
            if len(raw) != length:
                return self.respond(400, {"error": "incomplete_body"})
            payload = json.loads(raw)
            if not isinstance(payload, dict) or set(payload) != {"sku", "percent"}:
                return self.respond(400, {"error": "body_must_contain_sku_and_percent"})
            sku, percent = payload["sku"], payload["percent"]
            if sku not in PRICES:
                return self.respond(400, {"error": "unknown_sku"})
            if isinstance(percent, bool) or not isinstance(percent, (int, float)) or not 0 <= percent <= 100:
                return self.respond(400, {"error": "percent_must_be_0_to_100"})
        except (ValueError, UnicodeError):
            return self.respond(400, {"error": "invalid_json_body"})
        price = str((PRICES[sku] * (1 - Decimal(str(percent)) / 100)).quantize(CENT, ROUND_HALF_UP))
        with sqlite3.connect(self.server.database) as database:
            cursor = database.execute("INSERT INTO quotes(sku, percent, price) VALUES (?, ?, ?)", (sku, percent, price))
            quote_id = cursor.lastrowid
        self.respond(201, {"id": quote_id, "sku": sku, "percent": percent, "price": price, "currency": "USD"})

    def do_DELETE(self):
        path = urlsplit(self.path).path
        if not path.startswith(PREFIX) or not path[len(PREFIX):].isdigit():
            return self.respond(404, {"error": "route_not_found"})
        with sqlite3.connect(self.server.database) as database:
            cursor = database.execute("DELETE FROM quotes WHERE id = ?", (int(path[len(PREFIX):]),))
        self.respond(200 if cursor.rowcount else 404, {"withdrawn": cursor.rowcount == 1})


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--port", type=int, default=8087)
    parser.add_argument("--data", type=Path, default=Path(".pricing"))
    args = parser.parse_args()
    args.data.mkdir(mode=0o700, parents=True, exist_ok=True)
    database = args.data / "quotes.db"
    with sqlite3.connect(database) as connection:
        connection.execute("CREATE TABLE IF NOT EXISTS quotes(id INTEGER PRIMARY KEY, sku TEXT NOT NULL, percent REAL NOT NULL, price TEXT NOT NULL)")
    server = ThreadingHTTPServer(("127.0.0.1", args.port), Pricing)
    server.database = database
    print(f"Pricing API: http://127.0.0.1:{args.port}; data: {database}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()