用 Python 實作 RFC 10008 的 HTTP QUERY 方法：它和 GET、POST 差在哪

{
  "categories": ["laptops", "phones"],
  "price": {"min": 500, "max": 2000},
  "tags": ["pro"],
  "min_rating": 4.5,
  "in_stock": true,
  "near": {"lat": 35.68, "lng": 139.76, "radius_deg": 1.0},
  "sort": {"field": "price", "order": "desc"}
}

async def search_dispatcher(request: Request) -> JSONResponse:
    """Dispatch to a handler according to the HTTP method"""
    match request.method:
        case "GET":
            return await search_via_get(request)
        case "POST":
            return await search_via_post(request)
        case "QUERY":
            return await search_via_query(request)
        case "OPTIONS":
            return await product_search_options(request)
        case _:
            return JSONResponse(
                {"error": f"Method {request.method} not allowed"},
                status_code=405,
                headers={"Allow": "GET, POST, QUERY, OPTIONS"},
            )

routes = [
    Route(
        "/products/search",
        search_dispatcher,
        methods=["GET", "POST", "QUERY", "OPTIONS"],
    ),
]

app = Starlette(routes=routes)

async def search_via_get(request: Request) -> JSONResponse:
    params = request.query_params
    query: dict[str, Any] = {}

    if cats := params.get("categories"):
        query["categories"] = cats.split(",")
    if price_min := params.get("price_min"):
        query.setdefault("price", {})["min"] = int(price_min)
    if price_max := params.get("price_max"):
        query.setdefault("price", {})["max"] = int(price_max)
    # ... individual params like near_lat, near_lng, near_radius are needed

async def search_via_post(request: Request) -> JSONResponse:
    body = await request.body()
    content_type = request.headers.get("content-type", "")
    if "json" not in content_type:
        return JSONResponse(
            {"error": "Content-Type must be application/json"},
            status_code=415,
        )

    query = json.loads(body)
    data = search_products(query)
    return JSONResponse(data)

async def search_via_query(request: Request) -> JSONResponse:
    body = await request.body()

    # RFC 10008 §3: a Content-Type header is mandatory
    content_type = request.headers.get("content-type", "")
    if not content_type:
        return JSONResponse(
            {"error": "QUERY requests MUST include a Content-Type header (RFC 10008 §3)"},
            status_code=400,
        )

    # RFC 10008 §3: for an unsupported media type, respond 415 + notify supported types with Accept-Query
    if "json" not in content_type:
        return JSONResponse(
            {"error": f"Unsupported media type: {content_type}"},
            status_code=415,
            headers={"Accept-Query": '"application/json"'},
        )

    # RFC 10008 §4: a QUERY response is cacheable
    # include a hash of the request body in the cache key
    cache_key = _cache_key("QUERY", request.url.path, body)
    if cached := _get_cached(cache_key):
        return JSONResponse(cached, headers={"X-Cache": "HIT"})

    try:
        query = json.loads(body)
    except json.JSONDecodeError as e:
        # RFC 10008 §3: syntactically correct but semantically unprocessable → 422
        return JSONResponse(
            {"error": f"Unprocessable query content: {e}"},
            status_code=422,
        )

    data = search_products(query)
    _set_cache(cache_key, data)

    return JSONResponse(
        data,
        headers={
            "X-Cache": "MISS",
            "Accept-Query": '"application/json"',
        },
    )

def _cache_key(method: str, path: str, body: bytes) -> str:
    body_hash = hashlib.sha256(body).hexdigest()[:16]
    return f"{method}:{path}:{body_hash}"

HTTP/1.1 200 OK
x-search-method: QUERY
x-cache: MISS
x-cache-key: QUERY:/products/search:7f73fb16e7395e7d
accept-query: "application/json"
x-note: Safe + idempotent + cacheable + structured body (RFC 10008)

{"total":1,"offset":0,"limit":10,"results":[{"id":4,"name":"iPhone 16 Pro",...}]}

HTTP/1.1 200 OK
x-search-method: QUERY
x-cache: HIT
x-cache-key: QUERY:/products/search:7f73fb16e7395e7d

import httpx
import json

SEARCH_QUERY = {
    "categories": ["laptops", "phones"],
    "price": {"min": 500, "max": 2000},
    "tags": ["pro"],
}

with httpx.Client(base_url="http://localhost:8000") as client:
    # httpx supports a custom HTTP method via the request() method
    resp = client.request(
        "QUERY",
        "/products/search",
        content=json.dumps(SEARCH_QUERY),
        headers={"Content-Type": "application/json"},
    )
    print(resp.json())

GET /products/search?... (flat query string)
  → URL length: 212 chars

No Content-Type → 400: QUERY requests MUST include a Content-Type header (RFC 10008 §3)
Wrong Content-Type → 415: Unsupported media type: text/plain
  Accept-Query header: "application/json"
Malformed JSON → 422: Unprocessable query content: ...

Accept-Query: "application/json", application/sql;charset="UTF-8"

Access-Control-Allow-Methods: GET, POST, QUERY, OPTIONS