tessl/pypi-tornado
Tornado is a Python web framework and asynchronous networking library designed for applications requiring long-lived connections to many users.
http-client-server.mddocs/
HTTP Client and Server
HTTP client for making requests and HTTP server for handling incoming connections. Supports both synchronous and asynchronous operations with comprehensive HTTP feature support.
Capabilities
HTTP Client
Asynchronous and synchronous HTTP clients for making requests with support for all HTTP methods, headers, authentication, and response handling.
class AsyncHTTPClient:
"""Non-blocking HTTP client."""
@classmethod
def configure(cls, impl, **kwargs):
"""Configure HTTP client implementation."""
def initialize(self, **kwargs):
"""Initialize client."""
def close(self):
"""Close client and clean up resources."""
def fetch(self, request, callback=None, **kwargs):
"""
Fetch HTTP request asynchronously.
Args:
request: HTTPRequest object or URL string
callback: Callback function (if not using async/await)
**kwargs: Additional request parameters
Returns:
HTTPResponse object (when awaited)
"""
def fetch_impl(self, request, callback):
"""Implementation-specific fetch method."""
class HTTPClient:
"""Blocking HTTP client."""
def __init__(self, async_client_class=None, **kwargs):
"""Initialize blocking client."""
def close(self):
"""Close client."""
def fetch(self, request, **kwargs) -> HTTPResponse:
"""
Fetch HTTP request synchronously.
Args:
request: HTTPRequest object or URL string
**kwargs: Additional request parameters
Returns:
HTTPResponse object
Raises:
HTTPClientError: On HTTP errors
"""HTTP Request
Request object encapsulating all aspects of an HTTP request including URL, method, headers, body, and various options.
class HTTPRequest:
"""HTTP request object."""
def __init__(self,
url: str,
method: str = "GET",
headers=None,
body=None,
auth_username: str = None,
auth_password: str = None,
connect_timeout: float = None,
request_timeout: float = None,
if_modified_since=None,
follow_redirects: bool = None,
max_redirects: int = None,
user_agent: str = None,
use_gzip: bool = None,
network_interface: str = None,
streaming_callback=None,
header_callback=None,
prepare_curl_callback=None,
proxy_host: str = None,
proxy_port: int = None,
proxy_username: str = None,
proxy_password: str = None,
allow_nonstandard_methods: bool = None,
validate_cert: bool = None,
ca_certs: str = None,
allow_ipv6: bool = None,
client_key: str = None,
client_cert: str = None,
body_producer=None,
expect_100_continue: bool = False,
decompress_response: bool = None,
ssl_options=None):
"""
Initialize HTTP request.
Args:
url: Request URL
method: HTTP method (GET, POST, etc.)
headers: HTTP headers dict or HTTPHeaders
body: Request body (str or bytes)
auth_username: HTTP auth username
auth_password: HTTP auth password
connect_timeout: Connection timeout in seconds
request_timeout: Total request timeout in seconds
if_modified_since: datetime for conditional requests
follow_redirects: Whether to follow redirects
max_redirects: Maximum number of redirects
user_agent: User-Agent header value
use_gzip: Whether to use gzip compression
network_interface: Network interface to use
streaming_callback: Callback for streaming response body
header_callback: Callback for response headers
prepare_curl_callback: Callback to configure curl
proxy_host: Proxy server hostname
proxy_port: Proxy server port
proxy_username: Proxy authentication username
proxy_password: Proxy authentication password
allow_nonstandard_methods: Allow non-standard HTTP methods
validate_cert: Whether to validate SSL certificates
ca_certs: Path to CA certificate file
allow_ipv6: Whether to allow IPv6 addresses
client_key: Path to client private key
client_cert: Path to client certificate
body_producer: Callable to generate request body
expect_100_continue: Whether to use Expect: 100-continue
decompress_response: Whether to decompress response
ssl_options: SSL options dict
"""HTTP Response
Response object containing status, headers, body, and metadata from an HTTP response.
class HTTPResponse:
"""HTTP response object."""
def __init__(self, request, code, headers=None, buffer=None, effective_url=None, error=None, request_time=None, time_info=None, reason=None):
"""Initialize HTTP response."""
@property
def body(self) -> bytes:
"""Response body as bytes."""
@property
def headers(self):
"""Response headers."""
@property
def code(self) -> int:
"""HTTP status code."""
@property
def reason(self) -> str:
"""HTTP reason phrase."""
@property
def effective_url(self) -> str:
"""Final URL after redirects."""
def rethrow(self):
"""Re-raise HTTPClientError if response was an error."""HTTP Server
HTTP server for handling incoming requests with support for SSL, multiple processes, and custom connection handling.
class HTTPServer:
"""Multi-threaded, non-blocking HTTP server."""
def __init__(self, request_callback, no_keep_alive: bool = False, xheaders: bool = False, ssl_options=None, protocol: str = None, decompress_request: bool = False, chunk_size: int = None, max_header_size: int = None, idle_connection_timeout: float = None, body_timeout: float = None, max_body_size: int = None, max_buffer_size: int = None, trusted_downstream=None):
"""
Initialize HTTP server.
Args:
request_callback: Function to handle requests
no_keep_alive: Disable HTTP keep-alive
xheaders: Use X-Real-Ip and X-Forwarded-For headers
ssl_options: SSL configuration options
protocol: HTTP protocol class
decompress_request: Decompress request bodies
chunk_size: Chunk size for streaming
max_header_size: Maximum header size
idle_connection_timeout: Idle connection timeout
body_timeout: Request body timeout
max_body_size: Maximum request body size
max_buffer_size: Maximum buffer size
trusted_downstream: Trusted proxy addresses
"""
def listen(self, port: int, address: str = ""):
"""
Listen on given port and address.
Args:
port: Port number
address: IP address to bind (empty for all interfaces)
"""
def bind(self, port: int, address: str = None, family=socket.AF_INET, backlog: int = 128, flags=None, reuse_port: bool = False):
"""
Bind to port without starting server.
Args:
port: Port number
address: IP address to bind
family: Socket family
backlog: Listen backlog
flags: Additional socket flags
reuse_port: Enable SO_REUSEPORT
"""
def start(self, num_processes: int = 1):
"""
Start server with specified number of processes.
Args:
num_processes: Number of processes (1 for single-process)
"""
def stop(self):
"""Stop accepting new connections."""
def handle_stream(self, stream, address):
"""Handle incoming stream connection."""HTTP Utilities
Utility functions and classes for HTTP header handling, URL manipulation, and request/response parsing.
class HTTPHeaders:
"""Case-insensitive HTTP headers dictionary."""
def __init__(self, *args, **kwargs):
"""Initialize headers."""
def add(self, name: str, value: str):
"""Add header (allows multiple values)."""
def get(self, name: str, default=None) -> str:
"""Get header value."""
def get_list(self, name: str) -> List[str]:
"""Get all values for header."""
def get_all(self) -> List[Tuple[str, str]]:
"""Get all header name-value pairs."""
def parse_line(self, line: str):
"""Parse header line."""
@classmethod
def parse(cls, headers: str):
"""Parse headers from string."""
class HTTPServerRequest:
"""HTTP request object for server-side handling."""
def __init__(self, method: str = None, uri: str = None, version: str = "HTTP/1.0", headers=None, body=None, host: str = None, files=None, connection=None, start_line=None, server_connection=None):
"""Initialize server request."""
@property
def cookies(self) -> Dict[str, str]:
"""Request cookies."""
def supports_http_1_1(self) -> bool:
"""Check if request supports HTTP/1.1."""
def write(self, chunk: bytes, callback=None):
"""Write response chunk."""
def finish(self):
"""Finish response."""
def url_concat(url: str, args) -> str:
"""
Concatenate URL with query arguments.
Args:
url: Base URL
args: Query arguments dict or list of tuples
Returns:
URL with query string
"""
def parse_body_arguments(content_type: str, body: bytes, arguments: Dict[str, List[bytes]], files: Dict[str, List]):
"""
Parse form-encoded request body.
Args:
content_type: Content-Type header value
body: Request body bytes
arguments: Dictionary to store parsed arguments
files: Dictionary to store uploaded files
"""
def parse_multipart_form_data(boundary: bytes, data: bytes, arguments: Dict[str, List[bytes]], files: Dict[str, List]):
"""
Parse multipart form data.
Args:
boundary: Multipart boundary
data: Form data bytes
arguments: Dictionary to store parsed arguments
files: Dictionary to store uploaded files
"""
def format_timestamp(ts) -> str:
"""Format timestamp for HTTP headers."""
def parse_request_start_line(line: str):
"""Parse HTTP request start line."""
def parse_response_start_line(line: str):
"""Parse HTTP response start line."""
def encode_username_password(username: str, password: str) -> str:
"""Encode username and password for HTTP auth."""
def split_host_and_port(netloc: str) -> Tuple[str, int]:
"""Split host and port from netloc."""HTTP/1.x Protocol
Low-level HTTP/1.x protocol implementation for custom servers and advanced use cases.
class HTTP1Connection:
"""HTTP/1.x connection handler."""
def __init__(self, stream, is_client: bool, params=None, context=None):
"""Initialize HTTP/1.x connection."""
def read_response(self, delegate):
"""Read HTTP response."""
def read_message(self, delegate):
"""Read HTTP message."""
def write_headers(self, start_line, headers, chunk=None, callback=None):
"""Write response headers."""
def write(self, chunk: bytes, callback=None):
"""Write response body chunk."""
def finish(self):
"""Finish response."""
class HTTP1ConnectionParameters:
"""Parameters for HTTP/1.x connections."""
def __init__(self, no_keep_alive: bool = False, chunk_size: int = None, max_header_size: int = None, header_timeout: float = None, max_body_size: int = None, body_timeout: float = None, decompress: bool = False):
"""Initialize connection parameters."""Types
# HTTP method type
HTTPMethod = str
# Headers type
HTTPHeadersType = Union[HTTPHeaders, Dict[str, str]]
# Request callback type
HTTPRequestCallback = Callable[[HTTPRequest], None]
# Response callback type
HTTPResponseCallback = Callable[[HTTPResponse], None]
# Streaming callback type
StreamingCallback = Callable[[bytes], None]
# Header callback type
HeaderCallback = Callable[[str], None]Exceptions
class HTTPClientError(Exception):
"""Exception for HTTP client errors."""
def __init__(self, code: int, response=None):
"""
Initialize client error.
Args:
code: HTTP status code
response: HTTPResponse object
"""
class HTTPOutputError(Exception):
"""Exception for HTTP output errors."""
class HTTPInputError(Exception):
"""Exception for HTTP input parsing errors."""Install with Tessl CLI
npx tessl i tessl/pypi-tornado