tessl/pypi-starlette
The little ASGI library that shines.
requests-responses.mddocs/
Request & Response Handling
Starlette provides comprehensive request parsing and response generation capabilities for HTTP operations, including JSON, forms, files, streaming, and more.
HTTPConnection Base Class
from starlette.requests import HTTPConnection
from starlette.types import Scope, Receive
from typing import Mapping, Any, Iterator
class HTTPConnection(Mapping[str, Any]):
"""
Base class for incoming HTTP connections, providing functionality
common to both Request and WebSocket.
Provides access to ASGI scope data and connection information.
"""
def __init__(self, scope: Scope, receive: Receive | None = None) -> None:
"""
Initialize HTTP connection from ASGI scope.
Args:
scope: ASGI connection scope
receive: Optional ASGI receive callable
"""
# Mapping interface for scope access
def __getitem__(self, key: str) -> Any:
"""Access scope values by key."""
def __iter__(self) -> Iterator[str]:
"""Iterate over scope keys."""
def __len__(self) -> int:
"""Get number of scope items."""
# Connection properties
@property
def url(self) -> URL:
"""Connection URL."""
@property
def base_url(self) -> URL:
"""Base URL without query parameters."""
@property
def headers(self) -> Headers:
"""Connection headers."""
@property
def query_params(self) -> QueryParams:
"""URL query parameters."""
@property
def path_params(self) -> dict[str, Any]:
"""Path parameters from URL routing."""
@property
def cookies(self) -> dict[str, str]:
"""Request cookies."""
@property
def client(self) -> Address | None:
"""Client address (host, port)."""
@property
def session(self) -> dict[str, Any]:
"""Session data (requires SessionMiddleware)."""
@property
def auth(self) -> Any:
"""Authentication data (requires AuthenticationMiddleware)."""
@property
def user(self) -> Any:
"""Authenticated user (requires AuthenticationMiddleware)."""
@property
def state(self) -> State:
"""Connection-scoped state storage."""
def url_for(self, name: str, /, **path_params: Any) -> URL:
"""Generate URL for named route."""Request Class
from starlette.requests import Request, HTTPConnection, ClientDisconnect
from starlette.datastructures import URL, Headers, QueryParams, FormData, State, Address, MutableHeaders
from starlette.types import Scope, Receive, Send
from starlette._utils import AwaitableOrContextManager
from starlette.background import BackgroundTask
from datetime import datetime
from typing import Any, AsyncGenerator, Mapping, Literal
class Request(HTTPConnection):
"""
HTTP request object providing access to request data.
The Request class provides methods and properties for:
- HTTP method and URL information
- Headers and query parameters
- Request body parsing (JSON, forms, raw bytes)
- File uploads
- Cookies and sessions
- Client connection information
"""
def __init__(
self,
scope: Scope,
receive: Receive = empty_receive,
send: Send = empty_send
) -> None:
"""
Initialize request from ASGI scope.
Args:
scope: ASGI request scope
receive: ASGI receive callable
send: ASGI send callable (for push promises)
"""
# HTTP information
@property
def method(self) -> str:
"""HTTP method (GET, POST, etc.)."""
@property
def url(self) -> URL:
"""Full request URL including scheme, host, path, and query."""
@property
def base_url(self) -> URL:
"""Base URL (scheme + netloc) for generating absolute URLs."""
@property
def headers(self) -> Headers:
"""Request headers (case-insensitive)."""
@property
def query_params(self) -> QueryParams:
"""URL query parameters."""
@property
def path_params(self) -> dict[str, Any]:
"""Path parameters extracted from URL pattern."""
# Client information
@property
def cookies(self) -> dict[str, str]:
"""Request cookies."""
@property
def client(self) -> Address | None:
"""Client address (host, port)."""
# Extensions (require middleware)
@property
def session(self) -> dict[str, Any]:
"""Session data (requires SessionMiddleware)."""
@property
def auth(self) -> Any:
"""Authentication credentials (requires AuthenticationMiddleware)."""
@property
def user(self) -> Any:
"""Authenticated user object (requires AuthenticationMiddleware)."""
@property
def state(self) -> State:
"""Request-scoped state storage."""
@property
def receive(self) -> Receive:
"""ASGI receive callable."""
# Body parsing methods
def stream(self) -> AsyncGenerator[bytes, None]:
"""
Async generator yielding request body chunks.
Yields:
bytes: Raw body chunks as they arrive
"""
async def body(self) -> bytes:
"""
Get complete request body as bytes.
Returns:
bytes: Complete request body
"""
async def json(self) -> Any:
"""
Parse request body as JSON.
Returns:
Any: Parsed JSON data
Raises:
ValueError: If body is not valid JSON
"""
def form(
self,
*,
max_files: int | float = 1000,
max_fields: int | float = 1000,
max_part_size: int = 1024 * 1024,
) -> AwaitableOrContextManager[FormData]:
"""
Parse request body as form data (multipart or urlencoded).
Args:
max_files: Maximum number of file uploads
max_fields: Maximum number of form fields
max_part_size: Maximum size per multipart part
Returns:
FormData: Parsed form data with files and fields
"""
async def close(self) -> None:
"""Close any open file uploads in form data."""
async def is_disconnected(self) -> bool:
"""
Check if client has disconnected.
Returns:
bool: True if client disconnected
"""
async def send_push_promise(self, path: str) -> None:
"""
Send HTTP/2 server push promise.
Args:
path: Path to push to client
"""
def url_for(self, name: str, /, **path_params: Any) -> URL:
"""
Generate absolute URL for named route.
Args:
name: Route name
**path_params: Path parameter values
Returns:
URL: Absolute URL for the route
"""Response Classes
Base Response
from starlette.responses import Response
from starlette.background import BackgroundTask
from starlette.datastructures import MutableHeaders
from typing import Any, Iterable, Mapping
class Response:
"""
Base HTTP response class.
Provides the foundation for all HTTP responses with:
- Status code and headers management
- Content rendering
- Cookie handling
- Background task execution
"""
def __init__(
self,
content: Any = None,
status_code: int = 200,
headers: Mapping[str, str] | None = None,
media_type: str | None = None,
background: BackgroundTask | None = None,
) -> None:
"""
Initialize HTTP response.
Args:
content: Response content (rendered by subclass)
status_code: HTTP status code
headers: Response headers
media_type: Content-Type header value
background: Background task to execute after response
"""
@property
def status_code(self) -> int:
"""HTTP status code."""
@property
def headers(self) -> MutableHeaders:
"""Response headers (mutable)."""
@property
def media_type(self) -> str | None:
"""Content-Type media type."""
@property
def charset(self) -> str:
"""Character encoding (default: utf-8)."""
@property
def background(self) -> BackgroundTask | None:
"""Background task to execute."""
def render(self, content: Any) -> bytes | memoryview:
"""
Render content to bytes.
Override in subclasses for custom content rendering.
Args:
content: Content to render
Returns:
bytes | memoryview: Rendered content
"""
def init_headers(self, headers: Mapping[str, str] | None = None) -> None:
"""Initialize response headers."""
def set_cookie(
self,
key: str,
value: str = "",
max_age: int | None = None,
expires: datetime | str | int | None = None,
path: str | None = "/",
domain: str | None = None,
secure: bool = False,
httponly: bool = False,
samesite: Literal["lax", "strict", "none"] | None = "lax",
partitioned: bool = False,
) -> None:
"""
Set HTTP cookie.
Args:
key: Cookie name
value: Cookie value
max_age: Cookie lifetime in seconds
expires: Cookie expiration timestamp
path: Cookie path
domain: Cookie domain
secure: Only send over HTTPS
httponly: Prevent JavaScript access
samesite: SameSite policy ("strict", "lax", or "none")
partitioned: Partitioned cookie (Chrome extension)
"""
def delete_cookie(
self,
key: str,
path: str = "/",
domain: str | None = None,
secure: bool = False,
httponly: bool = False,
samesite: Literal["lax", "strict", "none"] | None = "lax",
) -> None:
"""
Delete HTTP cookie by setting it to expire.
Args:
key: Cookie name to delete
path: Cookie path (must match original)
domain: Cookie domain (must match original)
secure: Secure flag (must match original)
httponly: HttpOnly flag (must match original)
samesite: SameSite policy (must match original)
"""JSON Response
from starlette.responses import JSONResponse
import json
class JSONResponse(Response):
"""
JSON response with application/json media type.
Automatically serializes content to JSON and sets appropriate headers.
"""
media_type = "application/json"
def render(self, content: Any) -> bytes:
"""
Serialize content to JSON bytes.
Args:
content: JSON-serializable data
Returns:
bytes: JSON-encoded content
"""
return json.dumps(
content,
ensure_ascii=False,
allow_nan=False,
indent=None,
separators=(",", ":"),
).encode("utf-8")Text Responses
from starlette.responses import PlainTextResponse, HTMLResponse
class PlainTextResponse(Response):
"""Plain text response with text/plain media type."""
media_type = "text/plain"
class HTMLResponse(Response):
"""HTML response with text/html media type."""
media_type = "text/html"Redirect Response
from starlette.responses import RedirectResponse
class RedirectResponse(Response):
"""
HTTP redirect response.
Returns a redirect to another URL with appropriate status code.
"""
def __init__(
self,
url: str,
status_code: int = 307,
headers: Mapping[str, str] | None = None,
background: BackgroundTask | None = None,
) -> None:
"""
Initialize redirect response.
Args:
url: Target URL for redirect
status_code: HTTP redirect status (301, 302, 307, 308)
headers: Additional response headers
background: Background task to execute
"""Streaming Response
from starlette.responses import StreamingResponse
from typing import AsyncIterable, Iterable, Union
# Content stream type aliases
SyncContentStream = Iterable[str | bytes]
AsyncContentStream = AsyncIterable[str | bytes]
ContentStream = Union[AsyncContentStream, SyncContentStream]
class StreamingResponse(Response):
"""
Streaming HTTP response for large or generated content.
Streams content to client as it becomes available, useful for:
- Large files
- Real-time generated data
- Server-sent events
- Reducing memory usage
"""
def __init__(
self,
content: ContentStream,
status_code: int = 200,
headers: Mapping[str, str] | None = None,
media_type: str | None = None,
background: BackgroundTask | None = None,
) -> None:
"""
Initialize streaming response.
Args:
content: Iterable or async iterable yielding chunks
status_code: HTTP status code
headers: Response headers
media_type: Content-Type header
background: Background task to execute
"""
async def listen_for_disconnect(self, receive: Receive) -> None:
"""Listen for client disconnect during streaming."""
async def stream_response(self, send: Send) -> None:
"""Stream response content to client."""File Response
from starlette.responses import FileResponse
from os import PathLike
from typing import BinaryIO
class FileResponse(Response):
"""
File download response with automatic headers.
Efficiently serves files with proper Content-Type, Content-Length,
and caching headers.
"""
chunk_size = 64 * 1024 # 64KB chunks
def __init__(
self,
path: str | PathLike[str],
status_code: int = 200,
headers: Mapping[str, str] | None = None,
media_type: str | None = None,
background: BackgroundTask | None = None,
filename: str | None = None,
stat_result: os.stat_result | None = None,
method: str | None = None,
content_disposition_type: str = "attachment",
) -> None:
"""
Initialize file response.
Args:
path: File system path to serve
status_code: HTTP status code
headers: Additional headers
media_type: Override Content-Type detection
background: Background task to execute
filename: Override filename for Content-Disposition
stat_result: Pre-computed file stats for performance
method: HTTP method (affects HEAD handling)
content_disposition_type: "attachment" or "inline"
"""
def set_stat_headers(self, stat_result: os.stat_result) -> None:
"""Set Content-Length and Last-Modified headers from file stats."""
@staticmethod
def is_not_modified(
response_headers: Headers,
request_headers: Headers,
) -> bool:
"""Check if file has been modified based on headers."""Request Data Access
Basic Request Information
async def request_info(request: Request):
return JSONResponse({
"method": request.method,
"url": str(request.url),
"path": request.url.path,
"query": str(request.query_params),
"headers": dict(request.headers),
"client": request.client and request.client.host,
})Query Parameters
from starlette.datastructures import QueryParams
async def search_endpoint(request: Request):
# Access query parameters
query = request.query_params.get("q", "")
page = request.query_params.get("page", "1")
per_page = request.query_params.get("per_page", "10")
# Convert to appropriate types
try:
page = int(page)
per_page = int(per_page)
except ValueError:
return JSONResponse({"error": "Invalid parameters"}, status_code=400)
# Multiple values for same parameter
tags = request.query_params.getlist("tag") # ?tag=python&tag=web
return JSONResponse({
"query": query,
"page": page,
"per_page": per_page,
"tags": tags,
})Path Parameters
from starlette.routing import Route
async def user_posts(request: Request):
# Extract path parameters
user_id = int(request.path_params["user_id"])
# Optional parameters
post_id = request.path_params.get("post_id")
if post_id:
post_id = int(post_id)
return JSONResponse({
"user_id": user_id,
"post_id": post_id,
})
routes = [
Route("/users/{user_id:int}", user_posts),
Route("/users/{user_id:int}/posts/{post_id:int}", user_posts),
]Request Headers
async def header_info(request: Request):
# Case-insensitive header access
content_type = request.headers.get("content-type")
user_agent = request.headers.get("user-agent")
authorization = request.headers.get("authorization")
# Custom headers
api_key = request.headers.get("x-api-key")
# Get all values for header (rare)
accept_values = request.headers.getlist("accept")
return JSONResponse({
"content_type": content_type,
"user_agent": user_agent,
"has_auth": bool(authorization),
"api_key": bool(api_key),
"accept": accept_values,
})Cookies
async def cookie_info(request: Request):
# Access cookies
session_id = request.cookies.get("session_id")
preferences = request.cookies.get("preferences", "{}")
return JSONResponse({
"session_id": bool(session_id),
"preferences": preferences,
"all_cookies": list(request.cookies.keys()),
})Request Body Parsing
JSON Data
async def json_endpoint(request: Request):
try:
# Parse JSON body
data = await request.json()
# Process data
result = {
"received": data,
"type": type(data).__name__,
}
return JSONResponse(result)
except ValueError as e:
return JSONResponse(
{"error": "Invalid JSON", "details": str(e)},
status_code=400
)Form Data
async def form_endpoint(request: Request):
# Parse form data (application/x-www-form-urlencoded or multipart/form-data)
form = await request.form()
# Access form fields
name = form.get("name", "")
email = form.get("email", "")
# Multiple values
interests = form.getlist("interest") # Multiple checkboxes/select
# File uploads
avatar = form.get("avatar") # UploadFile object
response_data = {
"name": name,
"email": email,
"interests": interests,
}
# Handle file upload
if avatar and avatar.filename:
# Read file content
content = await avatar.read()
response_data["avatar"] = {
"filename": avatar.filename,
"content_type": avatar.content_type,
"size": len(content),
}
# Close file to free memory
await avatar.close()
# Close form to clean up any remaining files
await form.close()
return JSONResponse(response_data)File Uploads
from starlette.datastructures import UploadFile
import os
async def upload_endpoint(request: Request):
form = await request.form()
uploaded_files = []
# Process multiple file uploads
for field_name, file in form.items():
if isinstance(file, UploadFile) and file.filename:
# Save file to disk
file_path = f"uploads/{file.filename}"
os.makedirs("uploads", exist_ok=True)
with open(file_path, "wb") as f:
# Read in chunks to handle large files
while chunk := await file.read(1024):
f.write(chunk)
uploaded_files.append({
"field": field_name,
"filename": file.filename,
"content_type": file.content_type,
"size": file.size,
"path": file_path,
})
await file.close()
await form.close()
return JSONResponse({"uploaded_files": uploaded_files})Raw Body Data
async def raw_body_endpoint(request: Request):
# Get complete body as bytes
body = await request.body()
return JSONResponse({
"body_size": len(body),
"content_type": request.headers.get("content-type"),
})
async def streaming_body_endpoint(request: Request):
# Process body in chunks (memory efficient for large uploads)
chunks = []
total_size = 0
async for chunk in request.stream():
chunks.append(len(chunk))
total_size += len(chunk)
# Process chunk...
# Early termination if too large
if total_size > 10 * 1024 * 1024: # 10MB limit
return JSONResponse(
{"error": "Request too large"},
status_code=413
)
return JSONResponse({
"chunks": len(chunks),
"total_size": total_size,
})Response Generation
JSON Responses
async def api_endpoint(request: Request):
# Simple JSON response
return JSONResponse({"message": "success"})
async def api_with_status(request: Request):
# JSON with custom status code
return JSONResponse(
{"error": "Resource not found"},
status_code=404
)
async def api_with_headers(request: Request):
# JSON with custom headers
return JSONResponse(
{"data": "value"},
headers={"X-Custom-Header": "value"}
)Text and HTML Responses
async def text_endpoint(request: Request):
return PlainTextResponse("Hello, plain text!")
async def html_endpoint(request: Request):
html_content = """
<!DOCTYPE html>
<html>
<head><title>Hello</title></head>
<body><h1>Hello, HTML!</h1></body>
</html>
"""
return HTMLResponse(html_content)Redirects
async def redirect_endpoint(request: Request):
# Temporary redirect (307)
return RedirectResponse("/new-location")
async def permanent_redirect(request: Request):
# Permanent redirect (301)
return RedirectResponse("/new-location", status_code=301)
async def redirect_with_params(request: Request):
# Redirect to named route
user_id = request.path_params["user_id"]
redirect_url = request.url_for("user_profile", user_id=user_id)
return RedirectResponse(redirect_url)File Downloads
async def download_file(request: Request):
file_path = "documents/report.pdf"
# Simple file download
return FileResponse(file_path, filename="monthly_report.pdf")
async def download_with_headers(request: Request):
# File download with custom headers
return FileResponse(
"data/export.csv",
media_type="text/csv",
headers={"X-Custom-Header": "value"},
filename="data_export.csv"
)
async def inline_file(request: Request):
# Display file inline (not download)
return FileResponse(
"images/chart.png",
content_disposition_type="inline"
)Streaming Responses
import asyncio
async def stream_endpoint(request: Request):
# Stream generated data
async def generate_data():
for i in range(100):
yield f"data chunk {i}\n"
await asyncio.sleep(0.1) # Simulate processing
return StreamingResponse(
generate_data(),
media_type="text/plain"
)
async def csv_stream(request: Request):
# Stream CSV data
async def generate_csv():
yield "id,name,email\n"
# Simulate database streaming
for i in range(1000):
yield f"{i},User {i},user{i}@example.com\n"
if i % 100 == 0:
await asyncio.sleep(0.01) # Yield control
return StreamingResponse(
generate_csv(),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=users.csv"}
)
async def server_sent_events(request: Request):
# Server-Sent Events stream
async def event_stream():
counter = 0
while True:
# Check if client disconnected
if await request.is_disconnected():
break
# Send event
yield f"data: Event {counter}\n\n"
counter += 1
await asyncio.sleep(1)
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
}
)Cookies and Sessions
Setting Cookies
async def set_cookie_endpoint(request: Request):
response = JSONResponse({"message": "Cookie set"})
# Simple cookie
response.set_cookie("simple", "value")
# Cookie with options
response.set_cookie(
key="preferences",
value="dark_theme",
max_age=30 * 24 * 60 * 60, # 30 days
secure=True,
httponly=True,
samesite="strict"
)
return response
async def delete_cookie_endpoint(request: Request):
response = JSONResponse({"message": "Cookie deleted"})
response.delete_cookie("preferences")
return responseBackground Tasks
from starlette.background import BackgroundTask
def send_email(to: str, subject: str, body: str):
# Simulate sending email
print(f"Sending email to {to}: {subject}")
async def user_signup(request: Request):
data = await request.json()
# Create user immediately
user_id = create_user(data)
# Send welcome email in background
task = BackgroundTask(
send_email,
to=data["email"],
subject="Welcome!",
body="Thanks for signing up!"
)
return JSONResponse(
{"user_id": user_id, "message": "User created"},
background=task
)Starlette's request and response system provides comprehensive tools for handling HTTP communication with proper parsing, validation, and response generation capabilities.
Install with Tessl CLI
npx tessl i tessl/pypi-starlette