REST: The HTTP QUERY Method (RFC 10008)¶

QUERY /contacts HTTP/1.1
Host: example.org
Content-Type: application/x-www-form-urlencoded
Accept: application/json

select=surname,givenname,email&limit=10&match=%22email=*@example.*%22

HTTP/1.1 200 OK
Content-Type: application/json

[{"surname": "Smith", "givenname": "John", "email": "smith@example.org"}]

HEAD /contacts HTTP/1.1
Host: example.org

HTTP/1.1 200 OK
Content-Type: application/xhtml
Accept-Query: application/x-www-form-urlencoded, application/sql

HTTP/1.1 303 See Other
Content-Type: text/plain
Location: /contacts/stored-queries/42

See stored query at "/contacts/stored-queries/42".

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse

app = FastAPI()

SUPPORTED = "application/json"
ACCEPT_QUERY = {"Accept-Query": f'"{SUPPORTED}"'}

CONTACTS = [
    {"id": 1, "name": "John Smith", "email": "smith@example.org", "status": "active"},
    {"id": 2, "name": "Ann Lee", "email": "lee@example.com", "status": "blocked"},
]

@app.api_route("/contacts", methods=["QUERY"])
async def query_contacts(request: Request) -> Response:
    content_type = request.headers.get("content-type", "").split(";")[0].strip()
    if not content_type:
        return JSONResponse({"error": "Content-Type is required"}, status_code=400)
    if content_type != SUPPORTED:
        return Response(status_code=415, headers=ACCEPT_QUERY)

    try:
        query = await request.json()
    except ValueError:
        return JSONResponse({"error": "Body is not valid JSON"}, status_code=400)
    if not isinstance(query, dict) or set(query) - {"status", "limit"}:
        return JSONResponse({"error": "Unsupported query fields"}, status_code=422)

    rows = [c for c in CONTACTS if "status" not in query or c["status"] == query["status"]]
    return JSONResponse(rows[: query.get("limit", 100)], headers=ACCEPT_QUERY)

@app.head("/contacts")
async def contacts_capabilities() -> Response:
    """Lets clients discover QUERY support via Accept-Query."""
    return Response(headers=ACCEPT_QUERY)

from starlette.endpoints import HTTPEndpoint
from starlette.responses import JSONResponse

class ContactSearch(HTTPEndpoint):
    async def query(self, request):
        return JSONResponse({"received": await request.json()})

import httpx

with httpx.Client(base_url="https://api.example.org") as client:
    response = client.request("QUERY", "/contacts", json={"status": "active", "limit": 10})
    response.raise_for_status()
    print(response.json())
    print("Supported query formats:", response.headers.get("accept-query"))

curl -i -X QUERY https://api.example.org/contacts \
  -H 'Content-Type: application/json' \
  --data '{"status": "active", "limit": 10}'

import pytest
from fastapi.testclient import TestClient

from app import CONTACTS, app

client = TestClient(app)

def query(body=None, **kwargs):
    return client.request("QUERY", "/contacts", json=body, **kwargs)

def test_query_returns_filtered_results():
    response = query({"status": "active"})
    assert response.status_code == 200
    assert [c["id"] for c in response.json()] == [1]
    assert response.headers["accept-query"] == '"application/json"'

def test_query_is_safe():
    before = [dict(c) for c in CONTACTS]
    for _ in range(3):
        query({"status": "blocked"})
    assert CONTACTS == before

def test_different_bodies_give_different_results():
    active = query({"status": "active"}).json()
    blocked = query({"status": "blocked"}).json()
    assert active != blocked

def test_missing_content_type_is_rejected():
    response = client.request("QUERY", "/contacts", content=b'{"status": "active"}')
    assert response.status_code == 400

@pytest.mark.parametrize("content_type", ["application/sql", "text/plain"])
def test_unsupported_media_type_lists_accepted_formats(content_type):
    response = client.request(
        "QUERY", "/contacts", content=b"select 1", headers={"Content-Type": content_type}
    )
    assert response.status_code == 415
    assert "application/json" in response.headers["accept-query"]

def test_unprocessable_query_returns_422():
    assert query({"unknown_field": 1}).status_code == 422

def test_get_is_not_allowed_on_query_resource():
    response = client.get("/contacts")
    assert response.status_code == 405