The comprehensive WSGI web application library providing essential utilities and components for building Python web applications.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Comprehensive HTTP parsing and formatting utilities for headers, cookies, dates, ETags, cache control, and content negotiation. These low-level functions provide the foundation for HTTP protocol handling in Werkzeug.
Functions for parsing and formatting HTTP headers with proper escaping and encoding.
def quote_header_value(value, allow_token=True):
"""
Add double quotes around a header value if needed.
Parameters:
- value: The value to quote (will be converted to string)
- allow_token: If False, always quote even if value contains only token chars
Returns:
Quoted header value string
"""
def unquote_header_value(value):
"""
Remove quotes from a header value and unescape internal quotes.
Parameters:
- value: Quoted header value string
Returns:
Unquoted header value
"""
def parse_list_header(value):
"""
Parse a header that contains a comma-separated list.
Parameters:
- value: Header value string
Returns:
List of individual values
"""
def parse_dict_header(value):
"""
Parse a header that contains key=value pairs.
Parameters:
- value: Header value string with key=value pairs
Returns:
Dictionary of key-value pairs
"""
def parse_options_header(value):
"""
Parse a Content-Type-like header with main value and parameters.
Parameters:
- value: Header value string (e.g., "text/html; charset=utf-8")
Returns:
Tuple of (main_value, parameters_dict)
"""
def dump_options_header(header, options):
"""
Create a Content-Type-like header from main value and options.
Parameters:
- header: Main header value
- options: Dictionary of parameters
Returns:
Complete header string
"""
def dump_header(iterable):
"""
Dump an iterable into a header value.
Parameters:
- iterable: Dictionary or iterable to convert to header
Returns:
Header string representation
"""Functions for parsing and handling HTTP Accept headers with quality values.
def parse_accept_header(value, cls=None):
"""
Parse an Accept header into structured data.
Parameters:
- value: Accept header string
- cls: Accept class to use (Accept, MIMEAccept, LanguageAccept, CharsetAccept)
Returns:
Accept object with parsed values and quality scores
"""Functions for parsing and managing Cache-Control headers.
def parse_cache_control_header(value, on_update=None, cls=None):
"""
Parse Cache-Control header into structured object.
Parameters:
- value: Cache-Control header string
- on_update: Callback function for updates
- cls: Cache control class (RequestCacheControl or ResponseCacheControl)
Returns:
Cache control object with directive properties
"""Functions for parsing and formatting CSP headers.
def parse_csp_header(value, on_update=None, cls=None):
"""
Parse Content-Security-Policy header.
Parameters:
- value: CSP header string
- on_update: Callback function for updates
- cls: CSP class to use
Returns:
ContentSecurityPolicy object
"""
def dump_csp_header(header):
"""
Convert CSP object to header string.
Parameters:
- header: ContentSecurityPolicy object
Returns:
CSP header string
"""Functions for parsing set-like headers.
def parse_set_header(value, on_update=None, cls=None):
"""
Parse set-like headers (comma-separated values).
Parameters:
- value: Header value string
- on_update: Callback for updates
- cls: Set class to use
Returns:
HeaderSet object
"""Functions for generating, parsing, and validating ETags.
def generate_etag(data):
"""
Generate an ETag from data using SHA-1 hash.
Parameters:
- data: Raw data bytes to hash
Returns:
ETag string (quoted hash)
"""
def quote_etag(etag, weak=False):
"""
Quote an ETag value for use in headers.
Parameters:
- etag: ETag value to quote
- weak: Whether this is a weak ETag (prepends W/)
Returns:
Quoted ETag string
"""
def unquote_etag(etag):
"""
Unquote an ETag and determine if it's weak.
Parameters:
- etag: Quoted ETag string or None
Returns:
Tuple of (unquoted_etag, is_weak)
"""
def parse_etags(value):
"""
Parse ETags header (If-Match, If-None-Match).
Parameters:
- value: ETags header string
Returns:
ETags object containing strong/weak etags and star flag
"""
def is_resource_modified(environ, etag=None, data=None, last_modified=None, ignore_if_range=True):
"""
Check if a resource was modified based on conditional headers.
Parameters:
- environ: WSGI environment or Request object
- etag: Current ETag of the resource
- data: Raw data to generate ETag from
- last_modified: Last modified timestamp
- ignore_if_range: Whether to ignore If-Range header
Returns:
True if resource was modified (should send full response)
"""Functions for parsing and formatting HTTP dates.
def http_date(timestamp=None):
"""
Format a timestamp as HTTP date string.
Parameters:
- timestamp: Unix timestamp, datetime, or None for current time
Returns:
HTTP date string (RFC 7231 format)
"""
def parse_date(value):
"""
Parse HTTP date string to datetime object.
Parameters:
- value: HTTP date string
Returns:
datetime object or None if parsing failed
"""
def parse_age(value=None):
"""
Parse Age header value.
Parameters:
- value: Age header string
Returns:
timedelta object representing the age
"""
def dump_age(age=None):
"""
Format age as Age header string.
Parameters:
- age: timedelta, int (seconds), or None
Returns:
Age header string
"""Functions for parsing and creating HTTP cookies.
def parse_cookie(header, cls=None):
"""
Parse Cookie header into dictionary.
Parameters:
- header: Cookie header string
- cls: Dictionary class to use for result
Returns:
Dictionary of cookie name-value pairs
"""
def dump_cookie(key, value="", max_age=None, expires=None, path="/", domain=None, secure=False, httponly=False, charset="utf-8", sync_expires=True, max_size=4093, samesite=None):
"""
Create Set-Cookie header string.
Parameters:
- key: Cookie name
- value: Cookie value
- max_age: Maximum age in seconds
- expires: Expiration date (datetime or timestamp)
- path: Cookie path
- domain: Cookie domain
- secure: Only send over HTTPS
- httponly: Prevent JavaScript access
- charset: Character encoding for value
- sync_expires: Sync expires with max_age
- max_size: Maximum cookie size
- samesite: SameSite attribute ('Strict', 'Lax', 'None')
Returns:
Complete Set-Cookie header string
"""Functions for handling HTTP Range requests.
def parse_range_header(value, make_inclusive=True):
"""
Parse Range header for partial content requests.
Parameters:
- value: Range header string (e.g., "bytes=0-1023")
- make_inclusive: Whether to make ranges inclusive
Returns:
Range object with units and range specifications
"""
def parse_content_range_header(value, on_update=None, cls=None):
"""
Parse Content-Range header from partial content responses.
Parameters:
- value: Content-Range header string
- on_update: Callback for updates
- cls: ContentRange class to use
Returns:
ContentRange object
"""
def parse_if_range_header(value):
"""
Parse If-Range header (ETag or date).
Parameters:
- value: If-Range header string
Returns:
IfRange object containing ETag or date
"""
def is_byte_range_valid(start, stop, length):
"""
Check if a byte range is valid for given content length.
Parameters:
- start: Range start position
- stop: Range stop position
- length: Total content length
Returns:
True if range is valid
"""Functions for classifying and filtering HTTP headers.
def is_entity_header(header):
"""
Check if header is an entity header.
Entity headers describe the content of the message body.
Parameters:
- header: Header name
Returns:
True if header is an entity header
"""
def is_hop_by_hop_header(header):
"""
Check if header is a hop-by-hop header.
Hop-by-hop headers are meaningful only for a single connection.
Parameters:
- header: Header name
Returns:
True if header is hop-by-hop
"""
def remove_entity_headers(headers, allowed=None):
"""
Remove entity headers from header collection.
Parameters:
- headers: Headers object or list of (name, value) tuples
- allowed: Set of entity headers to keep
Returns:
New Headers object without entity headers
"""
def remove_hop_by_hop_headers(headers):
"""
Remove hop-by-hop headers from header collection in-place.
Parameters:
- headers: Headers object or list of (name, value) tuples
"""HTTP status codes and policy enums.
# HTTP status code mappings
HTTP_STATUS_CODES: dict[int, str] = {
100: "Continue",
200: "OK",
201: "Created",
204: "No Content",
301: "Moved Permanently",
302: "Found",
304: "Not Modified",
400: "Bad Request",
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
405: "Method Not Allowed",
500: "Internal Server Error",
# ... complete mapping available
}
class COEP(Enum):
"""Cross Origin Embedder Policy values."""
UNSAFE_NONE = "unsafe-none"
REQUIRE_CORP = "require-corp"
class COOP(Enum):
"""Cross Origin Opener Policy values."""
UNSAFE_NONE = "unsafe-none"
SAME_ORIGIN_ALLOW_POPUPS = "same-origin-allow-popups"
SAME_ORIGIN = "same-origin"from werkzeug.http import parse_options_header, dump_options_header, parse_accept_header
def handle_content_negotiation(request):
# Parse Content-Type header
content_type, params = parse_options_header(request.headers.get('Content-Type'))
charset = params.get('charset', 'utf-8')
print(f"Content type: {content_type}, Charset: {charset}")
# Output: Content type: application/json, Charset: utf-8
# Create Content-Type header
ct_header = dump_options_header('text/html', {'charset': 'utf-8'})
print(ct_header) # "text/html; charset=utf-8"
# Parse Accept header
accept = parse_accept_header(request.headers.get('Accept'))
best_match = accept.best_match(['text/html', 'application/json', 'text/plain'])
if best_match == 'application/json':
return create_json_response()
elif best_match == 'text/html':
return create_html_response()
else:
return create_plain_text_response()from werkzeug.http import generate_etag, is_resource_modified, quote_etag
from werkzeug.wrappers import Response
import json
def serve_with_etag(request, data):
# Generate ETag from data
data_bytes = json.dumps(data, sort_keys=True).encode('utf-8')
etag = generate_etag(data_bytes)
# Check if resource was modified
if not is_resource_modified(request.environ, etag=etag):
# Return 304 Not Modified
response = Response(status=304)
response.headers['ETag'] = quote_etag(etag)
return response
# Return full response with ETag
response = Response(json.dumps(data), mimetype='application/json')
response.headers['ETag'] = quote_etag(etag)
return responsefrom werkzeug.http import parse_cookie, dump_cookie
from werkzeug.wrappers import Request, Response
from datetime import datetime, timedelta
def session_management(environ, start_response):
request = Request(environ)
# Parse incoming cookies
cookies = parse_cookie(request.headers.get('Cookie', ''))
session_id = cookies.get('session_id')
response = Response("Session handling example")
if not session_id:
# Create new session
session_id = generate_session_id() # Your function
# Set secure session cookie
cookie_header = dump_cookie(
'session_id',
session_id,
max_age=3600, # 1 hour
path='/',
secure=True,
httponly=True,
samesite='Lax'
)
response.headers['Set-Cookie'] = cookie_header
# Set additional cookies
theme_cookie = dump_cookie(
'theme',
'dark',
expires=datetime.now() + timedelta(days=30),
path='/'
)
response.headers.add('Set-Cookie', theme_cookie)
return response(environ, start_response)from werkzeug.http import parse_range_header, is_byte_range_valid
from werkzeug.wrappers import Request, Response
import os
def serve_file_with_ranges(request, file_path):
file_size = os.path.getsize(file_path)
# Parse Range header
range_header = request.headers.get('Range')
if range_header:
try:
ranges = parse_range_header(range_header)
if ranges and ranges.ranges:
# Get first range
start, end = ranges.ranges[0]
# Handle open-ended ranges
if start is None:
start = max(0, file_size - end)
end = file_size
elif end is None or end >= file_size:
end = file_size
# Validate range
if is_byte_range_valid(start, end, file_size):
# Serve partial content
with open(file_path, 'rb') as f:
f.seek(start)
data = f.read(end - start)
response = Response(
data,
status=206, # Partial Content
headers={
'Content-Range': f'bytes {start}-{end-1}/{file_size}',
'Accept-Ranges': 'bytes',
'Content-Length': str(len(data))
}
)
return response
except (ValueError, IndexError):
pass # Invalid range, serve full file
# Serve full file
with open(file_path, 'rb') as f:
data = f.read()
response = Response(
data,
headers={
'Accept-Ranges': 'bytes',
'Content-Length': str(file_size)
}
)
return responsefrom werkzeug.http import parse_cache_control_header
from werkzeug.datastructures import ResponseCacheControl
from werkzeug.wrappers import Response
from datetime import datetime, timedelta
def handle_caching(request):
# Parse request cache control
cache_control = parse_cache_control_header(
request.headers.get('Cache-Control')
)
if cache_control.no_cache:
# Client doesn't want cached response
response = generate_fresh_response()
elif cache_control.max_age is not None:
# Client specifies max age
response = get_cached_response_if_fresh(cache_control.max_age)
else:
response = get_normal_response()
# Set response cache control
response.cache_control.public = True
response.cache_control.max_age = 3600 # 1 hour
response.cache_control.must_revalidate = True
# Set other caching headers
response.expires = datetime.utcnow() + timedelta(hours=1)
response.last_modified = datetime.utcnow()
return responsefrom werkzeug.http import http_date, parse_date, parse_age, dump_age
from datetime import datetime, timedelta
def date_header_examples():
# Current time as HTTP date
now_header = http_date()
print(now_header) # "Wed, 21 Oct 2015 07:28:00 GMT"
# Specific timestamp
timestamp = 1445412480
date_header = http_date(timestamp)
# Parse HTTP date
parsed = parse_date("Wed, 21 Oct 2015 07:28:00 GMT")
print(parsed) # datetime(2015, 10, 21, 7, 28, 0)
# Age header handling
age = timedelta(hours=2, minutes=30)
age_header = dump_age(age)
print(age_header) # "9000" (seconds)
parsed_age = parse_age("9000")
print(parsed_age) # timedelta(seconds=9000)from werkzeug.datastructures import ContentSecurityPolicy
from werkzeug.http import dump_csp_header, parse_csp_header
def setup_csp():
# Create CSP policy
csp = ContentSecurityPolicy()
csp.default_src = ["'self'"]
csp.script_src = ["'self'", "'unsafe-inline'", "https://cdn.example.com"]
csp.style_src = ["'self'", "'unsafe-inline'"]
csp.img_src = ["'self'", "data:", "https:"]
# Convert to header string
csp_header = dump_csp_header(csp)
# Parse existing CSP header
existing_csp = "default-src 'self'; script-src 'self' 'unsafe-inline'"
parsed_csp = parse_csp_header(existing_csp)
return csp_headerfrom werkzeug.http import (
quote_header_value, unquote_header_value,
remove_entity_headers, is_hop_by_hop_header
)
from werkzeug.datastructures import Headers
def header_processing():
# Quote/unquote header values
quoted = quote_header_value("filename with spaces.txt")
print(quoted) # '"filename with spaces.txt"'
unquoted = unquote_header_value(quoted)
print(unquoted) # "filename with spaces.txt"
# Header classification
print(is_hop_by_hop_header('Connection')) # True
print(is_hop_by_hop_header('Content-Type')) # False
# Filter headers for forwarding
headers = Headers([
('Content-Type', 'text/html'),
('Content-Length', '1234'),
('Connection', 'close'),
('X-Custom', 'value')
])
# Remove entity headers (for HEAD requests)
filtered = remove_entity_headers(headers, allowed={'Content-Type'})
# Result has Content-Type and X-Custom, but not Content-LengthLow-level utilities for parsing form data from HTTP requests, including multipart and URL-encoded data handling.
def parse_form_data(environ, stream_factory=None, max_form_memory_size=None, max_content_length=None, cls=None, silent=True, *, max_form_parts=None):
"""
Parse form data from WSGI environ.
Parameters:
- environ: WSGI environment dictionary
- stream_factory: Callable for creating file streams
- max_form_memory_size: Maximum memory for form data
- max_content_length: Maximum total content length
- cls: MultiDict class to use (default: MultiDict)
- silent: Don't raise on parsing errors
- max_form_parts: Maximum number of multipart parts
Returns:
Tuple of (stream, form_dict, files_dict)
"""
def default_stream_factory(total_content_length, content_type, filename, content_length=None):
"""
Default stream factory for file uploads.
Parameters:
- total_content_length: Total request content length
- content_type: Content type of the part
- filename: Filename if provided
- content_length: Content length of this part
Returns:
File-like object for storing upload data
"""
class FormDataParser:
def __init__(self, stream_factory=None, max_form_memory_size=None, max_content_length=None, cls=None, silent=True, *, max_form_parts=None):
"""
Advanced form data parser with configurable limits.
Parameters:
- stream_factory: Callable for creating file streams
- max_form_memory_size: Maximum memory for form data
- max_content_length: Maximum total content length
- cls: MultiDict class to use
- silent: Don't raise on parsing errors
- max_form_parts: Maximum number of multipart parts
"""
def parse_from_environ(self, environ):
"""
Parse form data from WSGI environment.
Parameters:
- environ: WSGI environment dictionary
Returns:
Tuple of (stream, form_dict, files_dict)
"""Install with Tessl CLI
npx tessl i tessl/pypi-werkzeug