orjson and msgspec: Serialization Wins for FastAPI Worker Fleets

# ✅ Drop-in ORJSONResponse
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from pydantic import BaseModel

app = FastAPI(default_response_class=ORJSONResponse)

class Item(BaseModel):
    id: str
    score: float
    tags: list[str]

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: str) -> Item:
    return Item(id=item_id, score=0.9, tags=["a", "b"])

# ✅ msgspec for the absolute hottest encoder path
import msgspec

class ItemStruct(msgspec.Struct):
    id: str
    score: float
    tags: list[str]

encoder = msgspec.json.Encoder()

def encode_item(item: ItemStruct) -> bytes:
    return encoder.encode(item)  # returns bytes — perfect for Response

# ❌ Declaring victory from dumps of {"a": 1} in a notebook
# ✅ Replay production response fixtures (PII-redacted) in CI bench

import time
import orjson
import json

def bench(fn, payload, n=2000):
    t0 = time.perf_counter()
    for _ in range(n):
        fn(payload)
    return (time.perf_counter() - t0) / n * 1000

payload = {"items": [{"id": str(i), "tags": ["x"] * 20} for i in range(200)]}
print("stdlib", bench(lambda p: json.dumps(p).encode(), payload))
print("orjson", bench(orjson.dumps, payload))