Request Data

from flask import Flask, request
from flask.wrappers import Request as FlaskRequest
from werkzeug.sansio.request import Request as SansIORequest
from werkzeug.wrappers import Request as WerkzeugRequest

assert FlaskRequest.__mro__[1] is WerkzeugRequest
assert WerkzeugRequest.__mro__[1] is SansIORequest

app = Flask(__name__)


@app.get("/probe")
def probe():
    real = request._get_current_object()
    return {
        "proxy_class": request.__class__.__name__,
        "real_class": type(real).__name__,
        "same_object_twice": request.args is request.args,
        "args_type": type(request.args).__name__,
    }


with app.test_client() as client:
    print(client.get("/probe?q=dune").get_json())

from flask import Flask, jsonify, request

app = Flask(__name__)

BOOKS = [
    {"id": 1, "title": "Dune", "tags": ["sf", "classic"]},
    {"id": 2, "title": "Dune Messiah", "tags": ["sf"]},
    {"id": 3, "title": "The Left Hand of Darkness", "tags": ["sf", "classic"]},
    {"id": 4, "title": "A Wizard of Earthsea", "tags": ["fantasy", "classic"]},
]
MAX_PER_PAGE = 50


@app.get("/books")
def list_books():
    q = request.args.get("q", "").strip().lower()
    tags = request.args.getlist("tag")
    page = request.args.get("page", 1, type=int)
    per_page = min(request.args.get("per_page", 2, type=int), MAX_PER_PAGE)

    found = [
        b for b in BOOKS
        if q in b["title"].lower() and all(t in b["tags"] for t in tags)
    ]
    start = (max(page, 1) - 1) * max(per_page, 1)
    return jsonify(total=len(found), page=page, items=found[start:start + per_page])


client = app.test_client()
print(client.get("/books?q=dune&tag=sf&tag=classic").get_json())
print(client.get("/books?page=2").get_json()["items"])

curl 'http://localhost:5000/books?q=dune&tag=sf&tag=classic'

{"items":[{"id":1,"tags":["sf","classic"],"title":"Dune"}],"page":1,"total":1}

from flask import Flask, abort, jsonify, request

app = Flask(__name__)


def int_arg(name: str, default: int, *, minimum: int = 1, maximum: int = 100) -> int:
    """Read an integer query parameter strictly."""
    raw = request.args.get(name)
    if raw is None or raw == "":
        return default
    if not (raw.isascii() and raw.isdecimal()):
        abort(400, description=f"{name} must be a non-negative integer")
    value = int(raw)
    if not minimum <= value <= maximum:
        abort(400, description=f"{name} must be between {minimum} and {maximum}")
    return value


@app.get("/books")
def books():
    return jsonify(page=int_arg("page", 1, maximum=10_000), per_page=int_arg("per_page", 20))


client = app.test_client()
print(client.get("/books").get_json())
print(client.get("/books?per_page=500").status_code)
print(client.get("/books?page=abc").status_code)
print(client.get("/books?page=%2B1").status_code)

from flask import Flask, abort, jsonify, request

app = Flask(__name__)
CATALOGUE: list[dict] = []


@app.post("/books/new")
def create_book_from_form():
    title = request.form.get("title", "").strip()
    author = request.form.get("author", "").strip()
    year = request.form.get("year", type=int)
    tags = request.form.getlist("tag")
    if not title or not author:
        abort(400, description="title and author are required")
    book = {"title": title, "author": author, "year": year, "tags": tags}
    CATALOGUE.append(book)
    return jsonify(book), 201


client = app.test_client()
resp = client.post(
    "/books/new",
    data={"title": "Dune", "author": "Frank Herbert", "year": "1965", "tag": ["sf", "classic"]},
)
print(resp.status_code, resp.get_json())
print(client.post("/books/new", data={"title": "No author"}).status_code)

curl -X POST http://localhost:5000/books/new \
  -d 'title=Dune' -d 'author=Frank Herbert' -d 'year=1965' -d 'tag=sf' -d 'tag=classic'

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.route("/echo", methods=["GET", "POST"])
def echo():
    return jsonify(
        values=request.values.to_dict(flat=False),
        form=request.form.to_dict(flat=False),
        kind=type(request.values).__name__,
    )


client = app.test_client()
print(client.post("/echo?title=from-url", data={"title": "from-form", "author": "Ada"}).get_json())
print(client.get("/echo?title=from-url", data={"title": "ignored"}).get_json())

import io

from flask import Flask, abort, jsonify, request
from werkzeug.utils import secure_filename

app = Flask(__name__)
ALLOWED = {".png", ".jpg", ".jpeg"}


@app.post("/books/<int:book_id>/cover")
def upload_cover(book_id: int):
    cover = request.files.get("cover")
    if cover is None or not cover.filename:
        abort(400, description="choose a cover image")
    safe_name = secure_filename(cover.filename)
    suffix = "." + safe_name.rsplit(".", 1)[-1].lower() if "." in safe_name else ""
    if suffix not in ALLOWED:
        abort(415, description="covers must be PNG or JPEG")
    head = cover.stream.read(8)  # a real service would sniff the magic bytes here
    return jsonify(
        book=book_id,
        stored_as=f"cover-{book_id}{suffix}",
        claimed_name=cover.filename,
        safe_name=safe_name,
        claimed_type=cover.mimetype,
        first_bytes=len(head),
        title=request.form.get("caption", ""),
    ), 201


client = app.test_client()
resp = client.post(
    "/books/7/cover",
    data={"caption": "First edition", "cover": (io.BytesIO(b"\x89PNG\r\n\x1a\n" + b"x" * 100), "../../etc/Dune cover.PNG")},
)
print(resp.status_code, resp.get_json())
print(client.post("/books/7/cover", data={"caption": "no file"}).status_code)

from flask import Flask, abort, jsonify, request

app = Flask(__name__)
BOOKS: dict[int, dict] = {}


def validate_book(payload) -> tuple[dict, list[str]]:
    errors: list[str] = []
    if not isinstance(payload, dict):
        return {}, ["the body must be a JSON object"]
    title = payload.get("title")
    pages = payload.get("pages")
    if not isinstance(title, str) or not title.strip():
        errors.append("title must be a non-empty string")
    if pages is not None and (not isinstance(pages, int) or isinstance(pages, bool) or pages < 1):
        errors.append("pages must be a positive integer")
    # copy only the fields we know: never pass the whole payload to a model
    return {"title": (title or "").strip(), "pages": pages}, errors


@app.post("/api/books")
def create_book():
    payload = request.get_json()
    book, errors = validate_book(payload)
    if errors:
        return jsonify(errors=errors), 422
    book["id"] = len(BOOKS) + 1
    BOOKS[book["id"]] = book
    return jsonify(book), 201


client = app.test_client()
print(client.post("/api/books", json={"title": "Dune", "pages": 412, "is_admin": True}).get_json())
print(client.post("/api/books", data="title=Dune").status_code)
print(client.post("/api/books", data="{broken", content_type="application/json").status_code)
print(client.post("/api/books", json=["Dune"]).get_json())
print(client.post("/api/books", json={"title": "", "pages": -3}).get_json())

curl -X POST http://localhost:5000/api/books \
  -H 'Content-Type: application/json' \
  -d '{"title": "Dune", "pages": 412}'

{"id":1,"pages":412,"title":"Dune"}

import hashlib
import hmac
import secrets

from flask import Flask, abort, jsonify, request

app = Flask(__name__)
# In production the secret comes from the environment or a secrets manager, never from source code.
app.config["ISBN_WEBHOOK_SECRET"] = secrets.token_hex(32)


@app.post("/webhooks/isbn-sync")
def isbn_sync():
    request.max_content_length = 64 * 1024  # this endpoint never needs more than 64 KiB
    secret = app.config.get("ISBN_WEBHOOK_SECRET")
    if not secret:
        abort(503, description="webhook secret is not configured")

    body = request.get_data()  # the exact bytes the sender signed
    sent = request.headers.get("X-Signature-256", "")
    expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sent, expected):
        abort(401)

    payload = request.get_json(force=True)  # reuses the cached bytes
    return jsonify(accepted=len(payload.get("isbns", [])))


client = app.test_client()
body = b'{"isbns": ["9780441013593", "9780345339683"]}'
good = "sha256=" + hmac.new(app.config["ISBN_WEBHOOK_SECRET"].encode(), body, hashlib.sha256).hexdigest()
print(client.post("/webhooks/isbn-sync", data=body, headers={"X-Signature-256": good}).get_json())
print(client.post("/webhooks/isbn-sync", data=body, headers={"X-Signature-256": "sha256=00"}).status_code)
print(client.post("/webhooks/isbn-sync", data=body + b" ", headers={"X-Signature-256": good}).status_code)

import base64
import re

from flask import Flask, jsonify, request

app = Flask(__name__)
REQUEST_ID = re.compile(r"[A-Za-z0-9._-]{1,64}")


@app.get("/api/ping")
def ping():
    rid = request.headers.get("X-Request-ID", "")
    if not REQUEST_ID.fullmatch(rid):
        rid = "generated-by-server"  # never echo an unchecked header into logs or responses
    auth = request.authorization
    best = request.accept_mimetypes.best_match(["application/json", "text/html"])
    return jsonify(
        request_id=rid,
        auth_type=auth.type if auth else None,
        user=auth.username if auth and auth.type == "basic" else None,
        token_present=bool(auth and auth.type == "bearer" and auth.token),
        negotiated=best,
        language=request.accept_languages.best_match(["es", "en"]),
        content_type=request.content_type,
    )


client = app.test_client()
basic = "Basic " + base64.b64encode(b"ada:not-a-real-password").decode()
print(client.get("/api/ping", headers={
    "X-Request-ID": "abc-123", "Authorization": basic,
    "Accept": "text/html;q=0.5, application/json", "Accept-Language": "es-ES,es;q=0.9,en;q=0.5",
}).get_json())
print(client.get("/api/ping", headers={"Authorization": "Bearer tok", "X-Request-ID": "bad id!"}).get_json())

with app.test_request_context("/api/ping"):
    for access in (lambda: request.headers["X-Missing"], lambda: request.args["q"]):
        try:
            access()
        except KeyError as exc:
            print(type(exc).__name__, getattr(exc, "code", None))

from flask import Flask, jsonify, request

app = Flask(__name__)
THEMES = {"light", "dark"}


@app.get("/preferences")
def preferences():
    theme = request.cookies.get("theme", "light")
    if theme not in THEMES:  # a cookie is client input like any other
        theme = "light"
    return jsonify(theme=theme, cookie_names=sorted(request.cookies))


client = app.test_client()
client.set_cookie("theme", "dark")
client.set_cookie("tour_done", "1")
print(client.get("/preferences").get_json())
print(app.test_client().get("/preferences", headers={"Cookie": "theme=<script>"}).get_json())

from werkzeug.datastructures import ImmutableMultiDict

args = ImmutableMultiDict([("tag", "sf"), ("tag", "classic"), ("n", "5"), ("n", "x")])

assert args["tag"] == "sf"
assert args.getlist("tag") == ["sf", "classic"]
assert args.get("n", type=int) == 5
assert args.get("missing", 7, type=int) == 7          # missing -> default
assert args.getlist("n", type=int) == [5]             # "x" silently dropped
assert dict(args) == {"tag": "sf", "n": "5"}          # duplicates lost
assert args.to_dict(flat=False) == {"tag": ["sf", "classic"], "n": ["5", "x"]}
assert list(args.items(multi=True))[:2] == [("tag", "sf"), ("tag", "classic")]

try:
    args.add("a", "b")
except TypeError as exc:
    print("immutable:", exc)

editable = args.copy()
editable.add("tag", "fantasy")
print(type(editable).__name__, editable.getlist("tag"))

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.get("/books/<int:book_id>")
def show_book(book_id: int):
    return jsonify(
        method=request.method, path=request.path, full_path=request.full_path,
        query=request.query_string.decode(), script_root=request.script_root,
        url=request.url, base_url=request.base_url, url_root=request.url_root,
        host_url=request.host_url, host=request.host,
        endpoint=request.endpoint, view_args=request.view_args, rule=str(request.url_rule),
    )


client = app.test_client()
data = client.get("/books/42?lang=es", base_url="http://books.example.com:8000/shelf").get_json()
for key, value in data.items():
    print(f"{key:12} {value}")

from flask import Flask, jsonify, request
from werkzeug.middleware.proxy_fix import ProxyFix


def build(app_name: str, trust_proxy: bool) -> Flask:
    app = Flask(app_name)

    @app.get("/whoami")
    def whoami():
        return jsonify(
            remote_addr=request.remote_addr, scheme=request.scheme,
            host=request.host, access_route=request.access_route,
        )

    if trust_proxy:
        # exactly one proxy in front of the app: trust one hop of each header
        app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
    return app


forwarded = {
    # the client forged its own first entry; the trusted proxy appended the real peer
    "X-Forwarded-For": "198.51.100.7, 203.0.113.9",
    "X-Forwarded-Proto": "https",
    "X-Forwarded-Host": "books.example.com",
}
print("direct :", build("direct", False).test_client().get("/whoami", headers=forwarded).get_json())
print("proxied:", build("proxied", True).test_client().get("/whoami", headers=forwarded).get_json())

from flask import Flask, request

app = Flask(__name__)
app.config["TRUSTED_HOSTS"] = ["books.example.com", ".example.org"]


@app.get("/")
def index():
    return request.host


client = app.test_client()
for host in ("books.example.com", "evil.test", "shop.example.org", "example.org"):
    print(host, client.get("/", headers={"Host": host}).status_code)

import io

from flask import Flask, jsonify, request
from werkzeug.exceptions import RequestEntityTooLarge

app = Flask(__name__)
app.config.update(MAX_CONTENT_LENGTH=1024, MAX_FORM_MEMORY_SIZE=100, MAX_FORM_PARTS=3)


@app.post("/api/notes")
def notes():
    return jsonify(size=len(request.get_data()))  # raw bytes: only MAX_CONTENT_LENGTH applies


@app.post("/api/forms")
def forms():
    return jsonify(fields=len(request.form), files=len(request.files))  # parses the body


@app.post("/books/<int:book_id>/cover")
def cover(book_id: int):
    # this one view accepts large images; it must set the limits before touching the body
    request.max_content_length = 8 * 1024 * 1024
    request.max_form_memory_size = 10_000
    return jsonify(fields=len(request.form), files=len(request.files))


@app.errorhandler(RequestEntityTooLarge)
def too_large(error):
    return jsonify(error="request body too large"), 413


client = app.test_client()
print("body 500 bytes   ", client.post("/api/notes", data=b"x" * 500).status_code)
print("body 2000 bytes  ", client.post("/api/notes", data=b"x" * 2000).status_code)
print("urlencoded field ", client.post("/api/forms", data={"caption": "x" * 500}).status_code)
upload = {"caption": "x" * 500, "cover": (io.BytesIO(b"x" * 5000), "c.png")}
print("multipart, global", client.post("/api/forms", data=upload).status_code)
upload = {"caption": "x" * 500, "cover": (io.BytesIO(b"x" * 5000), "c.png")}
print("multipart, view  ", client.post("/books/1/cover", data=upload).status_code)
four = {"a": "1", "b": "2", "c": "3", "d": "4", "f": (io.BytesIO(b"1"), "f.txt")}
print("five parts       ", client.post("/api/forms", data=four).status_code)

from flask import Flask, jsonify, request


class MethodOverrideMiddleware:
    allowed_methods = frozenset({"PUT", "PATCH", "DELETE"})

    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        if environ.get("REQUEST_METHOD") == "POST":
            override = environ.get("HTTP_X_HTTP_METHOD_OVERRIDE", "").upper()
            if override in self.allowed_methods:
                environ["REQUEST_METHOD"] = override
        return self.app(environ, start_response)


app = Flask(__name__)
app.wsgi_app = MethodOverrideMiddleware(app.wsgi_app)


@app.route("/api/books/<int:book_id>", methods=["GET", "POST", "DELETE"])
def book(book_id: int):
    return jsonify(method=request.method, id=book_id)


client = app.test_client()
print(client.post("/api/books/3", headers={"X-HTTP-Method-Override": "DELETE"}).get_json())
print(client.get("/api/books/3", headers={"X-HTTP-Method-Override": "DELETE"}).get_json())
print(client.post("/api/books/3").get_json())

import hashlib
import io

from flask import Flask, jsonify, request


class ChecksumStream:
    def __init__(self, stream):
        self._stream = stream
        self.digest = hashlib.sha256()

    def read(self, size=-1):
        chunk = self._stream.read(size)
        self.digest.update(chunk)
        return chunk

    def readline(self, size=-1):
        line = self._stream.readline(size)
        self.digest.update(line)
        return line


def start_checksum(req):
    """Wrap wsgi.input of this request. Call it before the body is read."""
    stream = ChecksumStream(req.environ["wsgi.input"])
    req.environ["wsgi.input"] = stream
    return stream.digest


app = Flask(__name__)


@app.post("/api/imports")
def import_books():
    digest = start_checksum(request)
    form = dict(request.form)                       # parses form fields ...
    uploaded = list(request.files)                  # ... and file parts
    request.get_data(parse_form_data=True)          # reads whatever is left (JSON, octet-stream)
    return jsonify(sha256=digest.hexdigest(), fields=form, files=uploaded)


client = app.test_client()
resp = client.post("/api/imports", data={"note": "weekly", "csv": (io.BytesIO(b"title\nDune\n"), "books.csv")})
print(resp.get_json())

# a body of exactly known bytes gives a digest that can be verified independently
payload = b'{"isbns": ["9780441013593"]}'
resp = client.post("/api/imports", data=payload, content_type="application/octet-stream")
print(resp.get_json()["sha256"] == hashlib.sha256(payload).hexdigest())

import re
from functools import cached_property

from flask import Flask, Request, jsonify, request
from werkzeug.exceptions import BadRequest, HTTPException, UnsupportedMediaType


class BookshelfRequest(Request):
    @cached_property
    def request_id(self) -> str:
        candidate = self.headers.get("X-Request-ID", "")
        return candidate if re.fullmatch(r"[A-Za-z0-9._-]{1,64}", candidate) else "none"

    def on_json_loading_failed(self, e):
        # Flask's default hides the decoding detail outside debug mode; an API usually wants a short, safe reason.
        if e is None:
            raise UnsupportedMediaType("send the body as application/json")
        raise BadRequest("the request body is not valid JSON")


app = Flask(__name__)
app.request_class = BookshelfRequest


@app.errorhandler(HTTPException)
def as_json(error):
    return jsonify(error=error.name, detail=error.description, request_id=request.request_id), error.code


@app.post("/api/books")
def create():
    return jsonify(received=request.get_json()), 201


client = app.test_client()
print(client.post("/api/books", json={"title": "Dune"}, headers={"X-Request-ID": "r-1"}).get_json())
print(client.post("/api/books", data="title=Dune").get_json())
print(client.post("/api/books", data="{oops", content_type="application/json").get_json())

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.post("/books/search")
def search():
    return jsonify(
        q=request.args.get("q"), tags=request.args.getlist("tag"),
        remote=request.remote_addr, secure=request.is_secure, body=request.get_json(silent=True),
    )


client = app.test_client()
resp = client.post(
    "/books/search",
    query_string={"q": "dune", "tag": ["sf", "classic"]},
    json={"limit": 5},
    base_url="https://books.example.com/",
    environ_overrides={"REMOTE_ADDR": "203.0.113.9"},
)
print(resp.get_json())

with app.test_request_context("/books/search?q=herbert", method="POST", json={"limit": 1}):
    print(request.args["q"], request.get_json())