starlette 1.4.1: applications.Starlette.add_exception_handler

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify starlette.applications.Starlette.add_exception_handler in pkg:pypi/starlette@1.4.1
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/starlette@1.4.1
Demonstrate these symbols/APIs:
  - starlette.applications.Starlette.add_exception_handler

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:e26c23f05e998754e33848691e0154fa55988f76a98628191745e0cd92be6b85","contract":["add_exception_handler catches custom exception classes raised by endpoints and transforms them into structured HTTP responses","add_exception_handler catches HTTP status codes such as 404 and returns custom JSON error responses"],"goal":"verify starlette.applications.Starlette.add_exception_handler in pkg:pypi/starlette@1.4.1","kind":"HOW","packages":["pkg:pypi/starlette@1.4.1"],"schemaVersion":1,"symbols":["starlette.applications.Starlette.add_exception_handler"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/starlette@1.4.1"],"schemaVersion":1,"subject":"pkg:pypi/starlette@1.4.1","symbols":["starlette.applications.Starlette.add_exception_handler"],"verifierAdapter":"python@1"}

anyio==4.14.2
idna==3.19
starlette==1.4.1
typing-extensions==4.16.0

"""Sample demonstrating custom and HTTP exception handling using Starlette.add_exception_handler."""

from typing import Any, Dict
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route


class ValidationError(Exception):
    """Custom application exception for validation errors."""

    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code
        super().__init__(message)


async def validation_error_handler(request: Request, exc: Exception) -> JSONResponse:
    """Handle custom ValidationError exceptions."""
    assert isinstance(exc, ValidationError)
    return JSONResponse(
        {"error": exc.message, "code": exc.code},
        status_code=422,
    )


async def not_found_handler(request: Request, exc: Exception) -> JSONResponse:
    """Handle HTTP 404 status code exceptions."""
    return JSONResponse(
        {"error": "Resource not found", "path": request.url.path},
        status_code=404,
    )


async def trigger_validation_error(request: Request) -> JSONResponse:
    """Endpoint that raises a custom ValidationError."""
    raise ValidationError("Invalid input parameters", "ERR_INVALID_PARAMS")


async def trigger_http_forbidden(request: Request) -> JSONResponse:
    """Endpoint that raises an HTTPException with status 403."""
    raise HTTPException(status_code=403, detail="Forbidden access")


async def success_endpoint(request: Request) -> JSONResponse:
    """Endpoint that returns a successful response."""
    return JSONResponse({"status": "success"})


def create_app() -> Starlette:
    """Create and configure Starlette application with exception handlers."""
    routes = [
        Route("/validate", trigger_validation_error, methods=["POST"]),
        Route("/forbidden", trigger_http_forbidden, methods=["GET"]),
        Route("/success", success_endpoint, methods=["GET"]),
    ]
    app = Starlette(routes=routes)

    # Register custom exception handler for ValidationError class
    app.add_exception_handler(ValidationError, validation_error_handler)

    # Register custom exception handler for HTTP 404 status code
    app.add_exception_handler(404, not_found_handler)

    return app


if __name__ == "__main__":
    app = create_app()
    print("Starlette application configured with add_exception_handler.")

{
  "schemaVersion": 1,
  "goal": "verify starlette.applications.Starlette.add_exception_handler in pkg:pypi/starlette@1.4.1",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/starlette@1.4.1"
  ],
  "symbols": [
    "starlette.applications.Starlette.add_exception_handler"
  ]
}

"""Contract verification tests for Starlette.add_exception_handler."""

import asyncio
import json
import os
import sys
from typing import Any, Dict, List, Optional, Tuple

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from starlette.applications import Starlette
from sample import create_app


async def invoke_asgi(
    app: Starlette,
    method: str,
    path: str,
    body: bytes = b"",
    headers: Optional[List[Tuple[bytes, bytes]]] = None,
) -> Tuple[int, Dict[bytes, bytes], bytes]:
    """Helper to invoke ASGI application offline without network dependencies."""
    if headers is None:
        headers = [(b"host", b"localhost"), (b"content-type", b"application/json")]

    scope = {
        "type": "http",
        "http_version": "1.1",
        "method": method,
        "path": path,
        "raw_path": path.encode("latin-1"),
        "query_string": b"",
        "headers": headers,
    }

    sent_messages: List[Dict[str, Any]] = []

    async def receive() -> Dict[str, Any]:
        return {"type": "http.request", "body": body, "more_body": False}

    async def send(message: Dict[str, Any]) -> None:
        sent_messages.append(message)

    await app(scope, receive, send)

    status_code = 0
    resp_headers: Dict[bytes, bytes] = {}
    response_body = b""

    for msg in sent_messages:
        if msg["type"] == "http.response.start":
            status_code = msg["status"]
            resp_headers = dict(msg.get("headers", []))
        elif msg["type"] == "http.response.body":
            response_body += msg.get("body", b"")

    return status_code, resp_headers, response_body


def test_custom_exception_handler() -> None:
    """Assertion 1: add_exception_handler catches custom exception classes raised by endpoints and transforms them into structured HTTP responses."""
    app = create_app()

    async def run_test() -> None:
        status, _, body = await invoke_asgi(app, "POST", "/validate")
        assert status == 422, f"Expected status 422, got {status}"
        data = json.loads(body.decode("utf-8"))
        assert data == {
            "error": "Invalid input parameters",
            "code": "ERR_INVALID_PARAMS",
        }, f"Unexpected response body: {data}"

    asyncio.run(run_test())


def test_status_code_exception_handler() -> None:
    """Assertion 2: add_exception_handler catches HTTP status codes such as 404 and returns custom JSON error responses."""
    app = create_app()

    async def run_test() -> None:
        status, _, body = await invoke_asgi(app, "GET", "/missing-path")
        assert status == 404, f"Expected status 404, got {status}"
        data = json.loads(body.decode("utf-8"))
        assert data == {
            "error": "Resource not found",
            "path": "/missing-path",
        }, f"Unexpected response body: {data}"

    asyncio.run(run_test())


def test_success_endpoint() -> None:
    """Verify normal requests continue to succeed without exception handler intervention."""
    app = create_app()

    async def run_test() -> None:
        status, _, body = await invoke_asgi(app, "GET", "/success")
        assert status == 200, f"Expected status 200, got {status}"
        data = json.loads(body.decode("utf-8"))
        assert data == {"status": "success"}, f"Unexpected response body: {data}"

    asyncio.run(run_test())


def main() -> None:
    test_custom_exception_handler()
    test_status_code_exception_handler()
    test_success_endpoint()
    print("Starlette add_exception_handler contract verification passed.")


if __name__ == "__main__":
    main()