A next generation HTTP client for Python 3 with HTTP/2 support, async/await capabilities, and requests-compatible API.
npx @tessl/cli install tessl/pypi-http3@0.6.0A next-generation HTTP client for Python 3 with modern features including HTTP/2 and HTTP/1.1 support, async/await capabilities, connection pooling, and a requests-compatible API. HTTP3 enables building robust HTTP client applications with comprehensive timeout control, streaming support, and full type annotations.
pip install http3import http3For high-level request functions:
from http3 import get, post, put, patch, delete, head, options, requestFor client classes:
from http3 import Client, AsyncClientimport http3
# Simple synchronous requests
response = http3.get('https://www.example.org/')
print(response.status_code) # 200
print(response.protocol) # 'HTTP/2'
print(response.text) # Response content
# POST with JSON data
response = http3.post(
'https://api.example.com/data',
json={'key': 'value'},
headers={'Authorization': 'Bearer token'}
)
# Using a client for session-like behavior
with http3.Client() as client:
response = client.get('https://www.example.org/')
response2 = client.post('https://www.example.org/api')Async usage:
import http3
import asyncio
async def main():
async with http3.AsyncClient() as client:
response = await client.get('https://www.example.org/')
print(response.status_code)
asyncio.run(main())HTTP3 uses a modular architecture built around these key components:
This design enables HTTP3 to provide both simple request functions and sophisticated client capabilities while supporting modern web protocols and async patterns.
Simple functions for making HTTP requests without managing client instances. These provide a requests-compatible API with support for all standard HTTP methods.
def get(url: URLTypes, *, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def post(url: URLTypes, *, data: RequestData = None, files: RequestFiles = None, json: typing.Any = None, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def put(url: URLTypes, *, data: RequestData = None, files: RequestFiles = None, json: typing.Any = None, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def patch(url: URLTypes, *, data: RequestData = None, files: RequestFiles = None, json: typing.Any = None, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def delete(url: URLTypes, *, data: RequestData = None, files: RequestFiles = None, json: typing.Any = None, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def head(url: URLTypes, *, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = False, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def options(url: URLTypes, *, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, timeout: TimeoutTypes = None) -> Response: ...
def request(method: str, url: URLTypes, *, params: QueryParamTypes = None, data: RequestData = None, files: RequestFiles = None, json: typing.Any = None, headers: HeaderTypes = None, cookies: CookieTypes = None, auth: AuthTypes = None, timeout: TimeoutTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = True, stream: bool = False) -> Response: ...Persistent client instances for session-like behavior, connection pooling, and advanced configuration. Supports both synchronous and asynchronous operations.
class Client:
def __init__(self, auth: AuthTypes = None, cookies: CookieTypes = None, verify: VerifyTypes = True, cert: CertTypes = None, timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG, pool_limits: PoolLimits = DEFAULT_POOL_LIMITS, max_redirects: int = DEFAULT_MAX_REDIRECTS, base_url: URLTypes = None, dispatch: Dispatcher = None, app: typing.Callable = None, raise_app_exceptions: bool = True, backend: ConcurrencyBackend = None): ...
def request(self, method: str, url: URLTypes, *, data: RequestData = None, files: RequestFiles = None, json: typing.Any = None, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = None, timeout: TimeoutTypes = None) -> Response: ...
def get(self, url, **kwargs): ...
def post(self, url, **kwargs): ...
# ... other HTTP methods
def close(self): ...
class AsyncClient:
def __init__(self, auth: AuthTypes = None, cookies: CookieTypes = None, verify: VerifyTypes = True, cert: CertTypes = None, timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG, pool_limits: PoolLimits = DEFAULT_POOL_LIMITS, max_redirects: int = DEFAULT_MAX_REDIRECTS, base_url: URLTypes = None, dispatch: AsyncDispatcher = None, app: typing.Callable = None, raise_app_exceptions: bool = True, backend: ConcurrencyBackend = None): ...
async def request(self, method: str, url: URLTypes, *, data: AsyncRequestData = None, files: RequestFiles = None, json: typing.Any = None, params: QueryParamTypes = None, headers: HeaderTypes = None, cookies: CookieTypes = None, stream: bool = False, auth: AuthTypes = None, allow_redirects: bool = True, cert: CertTypes = None, verify: VerifyTypes = None, timeout: TimeoutTypes = None) -> AsyncResponse: ...
async def get(self, url, **kwargs): ...
async def post(self, url, **kwargs): ...
# ... other HTTP methods
async def close(self): ...Comprehensive objects for handling HTTP requests and responses with full data access, streaming support, and type safety.
class Request:
def __init__(self, method, url, *, data=b"", params=None, headers=None): ...
@property
def method(self) -> str: ...
@property
def url(self) -> URL: ...
@property
def headers(self) -> Headers: ...
class Response:
@property
def status_code(self) -> int: ...
@property
def protocol(self) -> str: ...
@property
def headers(self) -> Headers: ...
@property
def text(self) -> str: ...
@property
def content(self) -> bytes: ...
def json(self) -> typing.Any: ...
def read(self) -> bytes: ...
def stream(self) -> typing.Iterator[bytes]: ...
class AsyncRequest:
def __init__(self, method, url, *, data=b"", params=None, headers=None): ...
# Similar interface to Request
class AsyncResponse:
# Async versions of Response methods
async def json(self) -> typing.Any: ...
async def read(self) -> bytes: ...
async def stream(self) -> typing.AsyncIterator[bytes]: ...Flexible configuration classes for timeouts, SSL settings, connection pools, and other client behavior customization.
class TimeoutConfig:
def __init__(self, timeout=None, *, connect_timeout=None, read_timeout=None, write_timeout=None): ...
class SSLConfig:
def __init__(self, *, cert=None, verify=True): ...
class PoolLimits:
def __init__(self, *, soft_limit=None, hard_limit=None, pool_timeout=None): ...URL handling, headers management, query parameters, cookies, and other data structures used throughout the HTTP client.
class URL:
def __init__(self, url, allow_relative=False, params=None): ...
@property
def scheme(self) -> str: ...
@property
def host(self) -> str: ...
@property
def port(self) -> int: ...
@property
def path(self) -> str: ...
@property
def query(self) -> str: ...
def join(self, relative_url) -> "URL": ...
class Headers:
def __getitem__(self, key: str) -> str: ...
def __setitem__(self, key: str, value: str): ...
def get(self, key: str, default=None): ...
class QueryParams:
def __init__(self, params=None): ...
# Mapping-like interface
class Cookies:
def __init__(self, cookies=None): ...
# MutableMapping interfaceComprehensive exception hierarchy for handling different types of HTTP errors, timeouts, and connection issues.
# Timeout exceptions
class Timeout(Exception): ...
class ConnectTimeout(Timeout): ...
class ReadTimeout(Timeout): ...
class WriteTimeout(Timeout): ...
class PoolTimeout(Timeout): ...
# HTTP exceptions
class ProtocolError(Exception): ...
class DecodingError(Exception): ...
class InvalidURL(Exception): ...
# Redirect exceptions
class TooManyRedirects(Exception): ...
class RedirectBodyUnavailable(Exception): ...
class RedirectLoop(Exception): ...
# Stream exceptions
class StreamConsumed(Exception): ...
class ResponseNotRead(Exception): ...
class ResponseClosed(Exception): ...Low-level components for advanced use cases including custom dispatchers, connection management, and protocol-specific interfaces.
class HTTPConnection:
def __init__(self, origin: typing.Union[str, Origin], verify: VerifyTypes = True, cert: CertTypes = None, timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG, backend: ConcurrencyBackend = None, release_func: typing.Optional[ReleaseCallback] = None): ...
async def send(self, request: AsyncRequest, verify: VerifyTypes = None, cert: CertTypes = None, timeout: TimeoutTypes = None) -> AsyncResponse: ...
async def connect(self, verify: VerifyTypes = None, cert: CertTypes = None, timeout: TimeoutTypes = None) -> None: ...
async def close(self) -> None: ...
@property
def is_http2(self) -> bool: ...
class ConnectionPool:
def __init__(self, *, verify: VerifyTypes = True, cert: CertTypes = None, timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG, pool_limits: PoolLimits = DEFAULT_POOL_LIMITS, backend: ConcurrencyBackend = None): ...
async def send(self, request: AsyncRequest, verify: VerifyTypes = None, cert: CertTypes = None, timeout: TimeoutTypes = None) -> AsyncResponse: ...
async def close(self) -> None: ...
# Interface classes for custom implementations
class AsyncDispatcher:
async def send(self, request: AsyncRequest, verify: VerifyTypes = None, cert: CertTypes = None, timeout: TimeoutTypes = None) -> AsyncResponse: ...
async def close(self) -> None: ...
class Dispatcher:
def send(self, request: Request, verify: VerifyTypes = None, cert: CertTypes = None, timeout: TimeoutTypes = None) -> Response: ...
def close(self) -> None: ...
class ConcurrencyBackend:
async def connect(self, hostname: str, port: int, ssl_context: typing.Optional[ssl.SSLContext], timeout: TimeoutConfig) -> typing.Tuple[BaseReader, BaseWriter, Protocol]: ...
def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore: ...
class AsyncioBackend(ConcurrencyBackend):
# Default asyncio implementation
pass# Type aliases for flexible input handling
URLTypes = Union[URL, str]
QueryParamTypes = Union[QueryParams, Mapping[str, str], List[Tuple[Any, Any]], str]
HeaderTypes = Union[Headers, Dict[AnyStr, AnyStr], List[Tuple[AnyStr, AnyStr]]]
CookieTypes = Union[Cookies, CookieJar, Dict[str, str]]
AuthTypes = Union[Tuple[Union[str, bytes], Union[str, bytes]], Callable[[AsyncRequest], AsyncRequest]]
RequestData = Union[dict, str, bytes, Iterator[bytes]]
AsyncRequestData = Union[dict, str, bytes, AsyncIterator[bytes]]
RequestFiles = Dict[str, Union[IO[AnyStr], Tuple[str, IO[AnyStr]], Tuple[str, IO[AnyStr], str]]]
TimeoutTypes = Union[float, Tuple[float, float, float], TimeoutConfig]
CertTypes = Union[str, Tuple[str, str]]
VerifyTypes = Union[str, bool]
# Constants
USER_AGENT: str # Default user agent string
# Default configuration values
DEFAULT_TIMEOUT_CONFIG = TimeoutConfig(timeout=5.0)
DEFAULT_POOL_LIMITS = PoolLimits(soft_limit=10, hard_limit=100, pool_timeout=5.0)
DEFAULT_MAX_REDIRECTS = 20
# Status codes enum and utilities
class StatusCode(IntEnum):
# HTTP status codes with utility methods
@classmethod
def get_reason_phrase(cls, value: int) -> str: ...
@classmethod
def is_redirect(cls, value: int) -> bool: ...
@classmethod
def is_client_error(cls, value: int) -> bool: ...
@classmethod
def is_server_error(cls, value: int) -> bool: ...
codes: StatusCode # Status codes lookup instance