Request & Response

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


def inspect(request):
    info = {
        "method": request.method,
        "path": request.path,
        "path_qs": request.path_qs,
        "query_string": request.query_string,
        "url": request.url,
        "host": request.host,
        "host_url": request.host_url,
        "application_url": request.application_url,
        "scheme": request.scheme,
        "content_type": request.content_type,
        "user_agent": request.user_agent,
    }
    return Response(json=info)


with Configurator() as config:
    config.add_route("inspect", "/shelves/inspect")
    config.add_view(inspect, route_name="inspect")
    app = config.make_wsgi_app()

res = TestApp(app).get("/shelves/inspect?tag=fiction&tag=poetry", headers={"User-Agent": "curl/8.7"})
print(res.json)
assert res.json["path"] == "/shelves/inspect"
assert res.json["query_string"] == "tag=fiction&tag=poetry"
assert res.json["url"] == "http://localhost/shelves/inspect?tag=fiction&tag=poetry"

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


def list_books(request):
    # request.GET: query string only, request.params: query string + form body
    tags = request.GET.getall("tag")
    try:
        limit = int(request.GET.get("limit", "20"))
    except ValueError:
        limit = 20
    limit = max(1, min(limit, 100))  # never trust a client-supplied page size
    return Response(json={"tags": tags, "limit": limit, "q": request.params.get("q")})


def create_shelf(request):
    name = request.POST.get("name", "").strip()
    if not name:
        return Response(json={"error": "name is required"}, status=422)
    return Response(json={"created": name}, status=201)


with Configurator() as config:
    config.add_route("books", "/books")
    config.add_route("shelves", "/shelves")
    config.add_view(list_books, route_name="books", request_method="GET")
    config.add_view(create_shelf, route_name="shelves", request_method="POST")
    app = config.make_wsgi_app()

client = TestApp(app)
print(client.get("/books?tag=fiction&tag=poetry&limit=500&q=dune").json)
print(client.post("/shelves", {"name": "Sci-fi"}).json)
res = client.post("/shelves", {"name": "  "}, status=422)
assert client.get("/books?tag=a&tag=b").json["tags"] == ["a", "b"]
assert client.get("/books?limit=500").json["limit"] == 100
assert res.json == {"error": "name is required"}

import json

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPBadRequest, HTTPUnsupportedMediaType
from webtest import TestApp


def add_book(request):
    if request.content_type != "application/json":
        raise HTTPUnsupportedMediaType("send application/json")
    try:
        payload = request.json_body
    except ValueError:  # json.JSONDecodeError and invalid UTF-8 both derive from ValueError
        raise HTTPBadRequest("request body is not valid JSON")
    if not isinstance(payload, dict) or not isinstance(payload.get("title"), str):
        raise HTTPBadRequest("a JSON object with a string title is required")
    return {"added": payload["title"]}


with Configurator() as config:
    config.add_route("add_book", "/shelves/{shelf_id}/books")
    config.add_view(add_book, route_name="add_book", request_method="POST", renderer="json")
    app = config.make_wsgi_app()

client = TestApp(app)
print(client.post_json("/shelves/7/books", {"title": "Dune"}).json)
assert client.post("/shelves/7/books", "{oops", content_type="application/json", status=400)
assert client.post("/shelves/7/books", "title=Dune", status=415)
assert client.post_json("/shelves/7/books", ["Dune"], status=400)

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


def whoami(request):
    theme = request.cookies.get("theme", "light")
    if theme not in {"light", "dark"}:  # a cookie is user input
        theme = "light"
    lang = request.accept_language.lookup(["en", "es"], default="en")
    return Response(json={"theme": theme, "lang": lang, "agent": request.headers.get("User-Agent")})


with Configurator() as config:
    config.add_route("whoami", "/whoami")
    config.add_view(whoami, route_name="whoami")
    app = config.make_wsgi_app()

headers = {"Cookie": "theme=dark; evil=1", "Accept-Language": "es-ES,es;q=0.9,en;q=0.5", "User-Agent": "demo"}
print(TestApp(app).get("/whoami", headers=headers).json)
assert TestApp(app).get("/whoami", headers={"Cookie": "theme=<script>"}).json["theme"] == "light"

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPNotAcceptable
from pyramid.response import Response
from webtest import TestApp


def export_shelf(request):
    offers = request.accept.acceptable_offers(["application/json", "text/csv"])
    if not offers:
        raise HTTPNotAcceptable("available: application/json, text/csv")
    media_type = offers[0][0]
    if media_type == "text/csv":
        response = Response("title,author\nDune,Herbert\n", content_type="text/csv")
    else:
        response = Response(json=[{"title": "Dune", "author": "Herbert"}])
    response.vary = ["Accept"]  # caches must key on the header the answer depends on
    return response


with Configurator() as config:
    config.add_route("export", "/shelves/1/export")
    config.add_view(export_shelf, route_name="export")
    app = config.make_wsgi_app()

client = TestApp(app)
res = client.get("/shelves/1/export", headers={"Accept": "text/csv;q=0.9, application/json;q=0.5"})
print(res.content_type, res.headers["Vary"])
assert res.content_type == "text/csv"
assert client.get("/shelves/1/export", headers={"Accept": "application/json"}).content_type == "application/json"
assert client.get("/shelves/1/export", headers={"Accept": "image/png"}, status=406)

import hashlib

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPRequestEntityTooLarge
from pyramid.response import Response
from webtest import TestApp

MAX_BYTES = 1024


def checksum(request):
    declared = request.content_length
    if declared is not None and declared > MAX_BYTES:
        raise HTTPRequestEntityTooLarge("body too large")
    digest, total = hashlib.sha256(), 0
    while chunk := request.body_file.read(256):
        total += len(chunk)
        if total > MAX_BYTES:  # Content-Length can be absent or lie
            raise HTTPRequestEntityTooLarge("body too large")
        digest.update(chunk)
    return Response(json={"bytes": total, "sha256": digest.hexdigest()})


with Configurator() as config:
    config.add_route("checksum", "/checksum")
    config.add_view(checksum, route_name="checksum", request_method="PUT")
    app = config.make_wsgi_app()

client = TestApp(app)
print(client.put("/checksum", b"x" * 300, content_type="application/octet-stream").json)
assert client.put("/checksum", b"x" * 5000, content_type="application/octet-stream", status=413)

import tempfile
import uuid
from pathlib import Path

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPBadRequest, HTTPRequestEntityTooLarge
from pyramid.response import Response
from webtest import TestApp, Upload

UPLOAD_DIR = Path(tempfile.mkdtemp())
ALLOWED = {"image/png": ".png", "image/jpeg": ".jpg"}
MAX_COVER = 512 * 1024


def upload_cover(request):
    field = request.POST.get("cover")
    if field is None or isinstance(field, str):
        raise HTTPBadRequest("a file field named cover is required")
    suffix = ALLOWED.get(field.type)  # the client-declared type; check the content too in real code
    if suffix is None:
        raise HTTPBadRequest("only PNG and JPEG covers are accepted")
    data = field.file.read(MAX_COVER + 1)
    if len(data) > MAX_COVER:
        raise HTTPRequestEntityTooLarge("cover is larger than 512 KiB")
    stored = UPLOAD_DIR / f"{uuid.uuid4().hex}{suffix}"  # never reuse field.filename as a path
    stored.write_bytes(data)
    return Response(json={"original": field.filename, "stored": stored.name, "bytes": len(data)}, status=201)


with Configurator() as config:
    config.add_route("cover", "/shelves/{shelf_id}/cover")
    config.add_view(upload_cover, route_name="cover", request_method="POST")
    app = config.make_wsgi_app()

client = TestApp(app)
res = client.post("/shelves/3/cover", upload_files=[("cover", "../../etc/passwd.png", b"\x89PNG\r\n")],
                  content_type="multipart/form-data")
print(res.json)
assert (UPLOAD_DIR / res.json["stored"]).exists() and ".." not in res.json["stored"]
assert client.post("/shelves/3/cover", {"title": "x"}, status=400)

from pyramid.response import Response

response = Response("welcome back")
response.set_cookie("theme", "dark", max_age=60 * 60 * 24 * 30, path="/", secure=True, httponly=True, samesite="Lax")
print(response.headers.getall("Set-Cookie"))

# delete_cookie ADDS an expiring Set-Cookie header; it does not replace an earlier one
response.delete_cookie("theme")
print(len(response.headers.getall("Set-Cookie")))  # 2

# to cancel a cookie that was set earlier in the same response, unset it
fresh = Response("x")
fresh.set_cookie("theme", "dark")
fresh.unset_cookie("theme")
assert fresh.headers.getall("Set-Cookie") == []

from urllib.parse import urlsplit

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPFound, HTTPSeeOther
from webtest import TestApp


def safe_next(request, default):
    target = request.params.get("next", "")
    parts = urlsplit(target)
    # only a plain local path: no scheme, no host, and not a protocol-relative "//host" URL
    if target.startswith("/") and not target.startswith("//") and not parts.scheme and not parts.netloc:
        return target
    return default


def login_done(request):
    return HTTPSeeOther(location=safe_next(request, request.route_url("home")))


def home(request):
    return HTTPFound(location="/shelves")


with Configurator() as config:
    config.add_route("home", "/")
    config.add_route("login_done", "/login/done")
    config.add_view(home, route_name="home")
    config.add_view(login_done, route_name="login_done")
    app = config.make_wsgi_app()

client = TestApp(app)
assert client.get("/login/done?next=/shelves/2").headers["Location"] == "http://localhost/shelves/2"
assert client.get("/login/done?next=https://evil.example/").headers["Location"] == "http://localhost/"
assert client.get("/login/done?next=//evil.example/").headers["Location"] == "http://localhost/"
print(client.get("/login/done").status, client.get("/login/done").headers["Location"])

from pyramid.config import Configurator
from webtest import TestApp


def create_shelf(request):
    shelf_id = 42
    request.response.status = 201
    request.response.headers["Location"] = request.route_url("shelf", shelf_id=shelf_id)
    request.response.set_cookie("last_shelf", str(shelf_id), samesite="Lax", httponly=True)
    return {"id": shelf_id, "name": request.POST.get("name")}


with Configurator() as config:
    config.add_route("shelf", "/shelves/{shelf_id}")
    config.add_route("shelves", "/shelves")
    config.add_view(create_shelf, route_name="shelves", request_method="POST", renderer="json")
    app = config.make_wsgi_app()

res = TestApp(app).post("/shelves", {"name": "Poetry"}, status=201)
print(res.headers["Location"], res.json)
assert res.headers["Location"] == "http://localhost/shelves/42"
assert "last_shelf=42" in res.headers["Set-Cookie"]

from dataclasses import dataclass

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


@dataclass
class Created:
    location: str


def created_adapter(result):
    return Response(status=201, headers={"Location": result.location})


def text_adapter(result):
    return Response(result, content_type="text/plain")


def add_book(request):
    return Created(location=request.route_url("book", book_id=9))


def ping(request):
    return "pong"


with Configurator() as config:
    config.add_response_adapter(created_adapter, Created)
    config.add_response_adapter(text_adapter, str)
    config.add_route("books", "/books")
    config.add_route("book", "/books/{book_id}")
    config.add_route("ping", "/ping")
    config.add_view(add_book, route_name="books", request_method="POST")
    config.add_view(ping, route_name="ping")
    app = config.make_wsgi_app()

client = TestApp(app)
print(client.post("/books").headers["Location"], client.get("/ping").text)
assert client.post("/books", status=201)
assert client.get("/ping").content_type == "text/plain"

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


def catalogue(request):
    return Response("public catalogue")


def my_shelves(request):
    return Response("private shelves")


def etagged(request):
    # a validator lets caches revalidate with If-None-Match and get a 304
    return Response("stable text", etag="shelf-v7", conditional_response=True)


def sensitive(request):
    response = Response("one-time data")
    response.cache_control.prevent_auto = True   # opt out of the automatic headers of http_cache
    response.cache_control.no_store = True
    return response


with Configurator() as config:
    for name in ("catalogue", "mine", "tagged", "sensitive"):
        config.add_route(name, "/" + name)
    config.add_view(catalogue, route_name="catalogue", http_cache=(3600, {"public": True}))
    config.add_view(my_shelves, route_name="mine", http_cache=0)
    config.add_view(etagged, route_name="tagged", http_cache=60)
    config.add_view(sensitive, route_name="sensitive", http_cache=3600)
    app = config.make_wsgi_app()

client = TestApp(app)
for path in ("/catalogue", "/mine", "/tagged", "/sensitive"):
    res = client.get(path)
    print(path, {k: v for k, v in res.headers.items() if k in ("Cache-Control", "ETag", "Pragma")})
assert client.get("/catalogue").headers["Cache-Control"] == "max-age=3600, public"
assert "no-store" in client.get("/mine").headers["Cache-Control"]
assert client.get("/sensitive").headers["Cache-Control"] == "no-store"
assert client.get("/tagged", headers={"If-None-Match": '"shelf-v7"'}).status_int == 304

import datetime

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp

UPDATED = datetime.datetime(2026, 9, 1, 12, 0, tzinfo=datetime.timezone.utc)
work = {"renders": 0}


def shelf(request):
    # cheap validator first: the last-modified time comes from a column, not from rendering
    response = Response(last_modified=UPDATED, conditional_response=True)
    response.cache_control = "private, max-age=0, must-revalidate"
    if request.if_modified_since and request.if_modified_since >= UPDATED:
        return response  # WebOb finishes the 304; no body work done
    work["renders"] += 1  # the expensive part
    response.text = "<h1>Poetry</h1>"
    return response


with Configurator() as config:
    config.add_route("shelf", "/shelves/1")
    config.add_view(shelf, route_name="shelf")
    app = config.make_wsgi_app()

client = TestApp(app)
first = client.get("/shelves/1")
print(first.headers["Last-Modified"], first.headers["Cache-Control"])
again = client.get("/shelves/1", headers={"If-Modified-Since": first.headers["Last-Modified"]})
print(again.status, again.body, work)
assert again.status_int == 304 and work["renders"] == 1