### Request Initialization Examples Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Demonstrates creating simple GET requests, POST requests with headers and JSON content, and streaming request bodies. ```python from pyqwest import Request, Headers # Simple GET request request = Request("GET", "https://httpbin.org/get") # POST with headers and JSON content headers = Headers({"Content-Type": "application/json"}) request = Request( "POST", "https://httpbin.org/post", headers=headers, content={"key": "value"}, params={"page": "1"} ) # Streaming request content async def stream_content(): yield b"chunk1" yield b"chunk2" request = Request( "POST", "https://httpbin.org/post", content=stream_content() ) ``` -------------------------------- ### Install pyqwest Source: https://github.com/curioswitch/pyqwest/blob/main/docs/index.md Installation commands for common Python package managers. ```bash uv add pyqwest ``` ```bash pip install pyqwest ``` -------------------------------- ### Perform HTTP GET requests Source: https://github.com/curioswitch/pyqwest/blob/main/docs/index.md Examples of making synchronous and asynchronous HTTP requests using pyqwest clients. ```python client = pyqwest.Client() response = await client.get("https://curioswitch.org") print(len(response.content)) ``` ```python client = pyqwest.SyncClient() response = client.get("https://curioswitch.org") print(len(response.content)) ``` -------------------------------- ### Perform Basic Async GET Request Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md Demonstrates a standard asynchronous GET request using the Client class. ```python import asyncio from pyqwest import Client async def main(): client = Client() response = await client.get("https://httpbin.org/get") print(response.status) print(response.text()) asyncio.run(main()) ``` -------------------------------- ### Initialize and use the Client Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Demonstrates creating a client with default or custom transport settings and performing a basic GET request. ```python import asyncio from pyqwest import Client, HTTPTransport async def main(): # Using default transport client = Client() # Using custom transport transport = HTTPTransport(timeout=10.0, enable_gzip=True) client = Client(transport=transport) # Make requests response = await client.get("https://httpbin.org/get") print(response.status) print(response.text()) asyncio.run(main()) ``` -------------------------------- ### Configure SyncHTTPTransport instance Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/transport.md Example of initializing the transport with custom timeout and cookie store settings. ```python from pyqwest import SyncHTTPTransport # Create transport with custom settings transport = SyncHTTPTransport( timeout=10.0, enable_cookie_store=True, ) ``` -------------------------------- ### Perform an Async HTTP Request Source: https://github.com/curioswitch/pyqwest/blob/main/README.md Initialize an async Client and use it to perform a GET request. ```python client = pyqwest.Client() response = await client.get("https://curioswitch.org") print(len(response.content)) ``` -------------------------------- ### Tune performance and connection settings Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/configuration.md Examples for adjusting connection pooling, disabling automatic decompression, and managing TCP keepalive intervals. ```python # Aggressive connection pooling transport = HTTPTransport( pool_max_idle_per_host=10, pool_idle_timeout=120.0, tcp_keepalive_interval=15.0, ) # No decompression (saves CPU) transport = HTTPTransport( enable_gzip=False, enable_brotli=False, enable_zstd=False, ) # Disable keepalive (for short-lived clients) transport = HTTPTransport(tcp_keepalive_interval=None) ``` -------------------------------- ### Process Streaming Response Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Full example demonstrating status reading, content streaming, and trailer access. ```python async def process_response(): response = await transport.execute(request) # Read status and headers print(f"Status: {response.status}") print(f"HTTP Version: {response.http_version}") # Stream content chunks = [] async for chunk in response.content: chunks.append(chunk) # Trailers now available (if using HTTP/2+) print(response.trailers) # Clean up if not fully consumed await response.aclose() ``` -------------------------------- ### Perform HTTP Requests Source: https://github.com/curioswitch/pyqwest/blob/main/docs/usage.md Execute GET and POST requests using the client instance. ```python response = await client.get("https://pyqwest.dev") assert response.status == 200 print(response.text()) response = await client.post( "https://httpbingo.org/post", headers={"content-type": "application/text", "user-agent": "pyqwest"}, content=b"Hello world!", ) print(response.text()) ``` ```python response = client.get("https://pyqwest.dev") assert response.status == 200 print(response.text()) response = client.post( "https://httpbingo.org/post", headers={"content-type": "application/text", "user-agent": "pyqwest"}, content=b"Hello world!", ) print(response.text()) ``` -------------------------------- ### Initialize RetryTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/retry-middleware.md Basic setup for wrapping a base transport with retry logic. ```python from pyqwest import Client, HTTPTransport from pyqwest.middleware.retry import RetryTransport # Create base transport base_transport = HTTPTransport(timeout=10.0) # Wrap with retry retry_transport = RetryTransport( base_transport, initial_interval=1.0, max_retries=5, ) client = Client(transport=retry_transport) response = await client.get("https://api.example.com/endpoint") ``` -------------------------------- ### Execute a GET request Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Performs a GET request with query parameters, including support for key-only parameters. ```python response = await client.get( "https://httpbin.org/get", params={"key": "value", "empty": None} ) ``` -------------------------------- ### Client.get Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes an asynchronous GET HTTP request. ```APIDOC ## async def get(url, headers=None, *, params=None) ### Description Executes a GET HTTP request. ### Parameters - **url** (str) - Required - The unencoded request URL. - **headers** (Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None) - Optional - The request headers. - **params** (dict[str, str | None] | Iterable[tuple[str, str | None]] | None) - Optional - Query parameters to append to the URL. ### Returns - **FullResponse** - The fully buffered HTTP response. ``` -------------------------------- ### Use SyncHTTPTransport as context manager Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/transport.md Example of managing transport lifecycle and executing a request using the context manager pattern. ```python with SyncHTTPTransport() as transport: request = SyncRequest("GET", "https://example.com") response = transport.execute_sync(request) ``` -------------------------------- ### Aggressive Retry Configuration Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/retry-middleware.md Example configuration for high-frequency, aggressive retry behavior. ```python RetryTransport( transport, initial_interval=0.1, # Start at 100ms randomization_factor=0.1, # Less jitter multiplier=2.0, # Double each time max_interval=5.0, # Cap at 5s max_retries=10, # Many retries ) ``` -------------------------------- ### HTTPTransport Usage Examples Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/transport.md Common patterns for initializing HTTPTransport, including default settings, custom TLS, specific HTTP versions, and OpenTelemetry integration. ```python from pyqwest import HTTPTransport, HTTPVersion # Basic transport with defaults transport = HTTPTransport() # Custom TLS transport = HTTPTransport( tls_ca_cert=open("ca.pem", "rb").read(), tls_cert=open("client.pem", "rb").read(), tls_key=open("client-key.pem", "rb").read(), ) # Force HTTP/2, set timeouts transport = HTTPTransport( http_version=HTTPVersion.HTTP2, timeout=30.0, connect_timeout=10.0, ) # With OpenTelemetry from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.trace import TracerProvider meter_provider = MeterProvider() tracer_provider = TracerProvider() transport = HTTPTransport( meter_provider=meter_provider, tracer_provider=tracer_provider, ) ``` -------------------------------- ### Execute GET Request Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Perform a synchronous GET request and access the response status. ```python response = client.get("https://httpbin.org/get", timeout=5.0) print(response.status) ``` -------------------------------- ### Use HTTPVersion in transport and responses Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/types.md Examples for specifying the HTTP version in transport constructors and reading it from responses. ```python from pyqwest import HTTPTransport, HTTPVersion # Force HTTP/2 transport = HTTPTransport(http_version=HTTPVersion.HTTP2) ``` ```python response = await client.get("https://example.com") print(response.http_version) # HTTPVersion.HTTP2 ``` -------------------------------- ### SyncClient.get Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes a synchronous GET HTTP request. ```APIDOC ## SyncClient.get ### Description Executes a GET HTTP request and returns a fully buffered FullResponse. ### Parameters - **url** (str) - Required - The unencoded request URL. - **headers** (Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None) - Optional - The request headers. - **timeout** (float | None) - Optional - Request timeout in seconds. - **params** (dict[str, str | None] | Iterable[tuple[str, str | None]] | None) - Optional - Query parameters. ### Response - **FullResponse** - The fully buffered HTTP response. ``` -------------------------------- ### Execute OPTIONS Request Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes an OPTIONS HTTP request. Parameters and behavior match get(). ```python async def options( url: str, headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None = None, *, params: dict[str, str | None] | Iterable[tuple[str, str | None]] | None = None, ) -> FullResponse ``` -------------------------------- ### Use Response as Context Manager Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Example of using the response within an async context manager to ensure automatic closure. ```python async with response: async for chunk in response.content: # Process chunk pass # Response automatically closed ``` -------------------------------- ### Get all header values Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/headers.md Retrieves all values for a specific header name, including duplicates. ```python cookies = headers.getall("Set-Cookie") for cookie in cookies: print(cookie) ``` -------------------------------- ### Execute HEAD Request Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes a HEAD HTTP request. Parameters and behavior match get(). ```python async def head( url: str, headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None = None, *, params: dict[str, str | None] | Iterable[tuple[str, str | None]] | None = None, ) -> FullResponse ``` -------------------------------- ### Perform Sync GET Request with Timeout Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md Uses the SyncClient for blocking requests with a specified timeout. ```python from pyqwest import SyncClient client = SyncClient() response = client.get( "https://api.example.com/data", timeout=10.0 ) print(response.status) ``` -------------------------------- ### HTTPTransport Context Manager Usage Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/transport.md Example of using HTTPTransport as an asynchronous context manager to handle resource lifecycle automatically. ```python async with HTTPTransport() as transport: request = Request("GET", "https://example.com") response = await transport.execute(request) ``` -------------------------------- ### Initialize Client Source: https://github.com/curioswitch/pyqwest/blob/main/docs/usage.md Create a client instance for either asynchronous or synchronous operations. ```python from pyqwest import Client client = Client() ``` ```python from pyqwest import SyncClient client = SyncClient() ``` -------------------------------- ### Configure Sync HTTPX Client with Pyqwest Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Shows how to set up a synchronous HTTPX client with compression and automatic cookie management via Pyqwest. ```python import httpx from pyqwest import SyncHTTPTransport from pyqwest.httpx import PyqwestTransport from pyqwest.middleware.retry import SyncRetryTransport def main(): # Configure with compression and cookies base_transport = SyncHTTPTransport( enable_gzip=True, enable_brotli=True, enable_cookie_store=True, timeout=10.0, ) # Add retry logic retry_transport = SyncRetryTransport(base_transport, max_retries=2) # Wrap for httpx httpx_transport = PyqwestTransport(retry_transport) # Use with httpx with httpx.Client(transport=httpx_transport) as client: # Cookies automatically managed response = client.get("https://api.example.com/login") # Subsequent requests include cookies protected = client.get("https://api.example.com/dashboard") return protected.json() data = main() ``` -------------------------------- ### Client Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Initializes a new asynchronous HTTP client instance. ```APIDOC ## Client(transport: Transport | None = None) ### Description Creates a new asynchronous HTTP client. ### Parameters - **transport** (Transport | None) - Optional - The transport to use for requests. If None, the shared default transport is used. ``` -------------------------------- ### Configure basic HTTPTransport settings Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/configuration.md Demonstrates initializing the transport with default settings, custom timeouts, and forcing a specific HTTP version. ```python from pyqwest import HTTPTransport # Minimal configuration (uses all defaults) transport = HTTPTransport() # Custom timeout transport = HTTPTransport(timeout=30.0, connect_timeout=10.0) # Force HTTP/2 from pyqwest import HTTPVersion transport = HTTPTransport(http_version=HTTPVersion.HTTP2) ``` -------------------------------- ### Initialize SyncClient Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Instantiate the client with either the default transport or a custom SyncHTTPTransport configuration. ```python from pyqwest import SyncClient, SyncHTTPTransport # Using default transport client = SyncClient() # Using custom transport transport = SyncHTTPTransport(timeout=10.0) client = SyncClient(transport=transport) ``` -------------------------------- ### SyncClient Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Initializes a new synchronous HTTP client instance. ```APIDOC ## SyncClient Constructor ### Description Creates a new synchronous HTTP client. If no transport is provided, the shared default transport is used. ### Parameters #### Arguments - **transport** (SyncTransport | None) - Optional - The transport to use for requests. ``` -------------------------------- ### Usage of SyncRetryTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/retry-middleware.md Example of wrapping a synchronous transport with retry logic and using it in a client. ```python from pyqwest import SyncClient, SyncHTTPTransport from pyqwest.middleware.retry import SyncRetryTransport base_transport = SyncHTTPTransport() retry_transport = SyncRetryTransport(base_transport, max_retries=3) client = SyncClient(transport=retry_transport) response = client.get("https://api.example.com/data") ``` -------------------------------- ### Handle Retry Exceptions Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/retry-middleware.md Example of catching ReadError when maximum retry attempts are exhausted. ```python from pyqwest import Client, ReadError from pyqwest.middleware.retry import RetryTransport retry_transport = RetryTransport(HTTPTransport(), max_retries=3) client = Client(transport=retry_transport) try: # Automatically retries on 5xx, 429, timeouts response = await client.get("https://api.example.com/data") except ReadError as e: print(f"Failed after retries: {e}") ``` -------------------------------- ### Handle StreamError exceptions Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/types.md Example of checking the error code within a StreamError exception. ```python from pyqwest import StreamError, StreamErrorCode try: response = await client.get("https://example.com") except StreamError as e: if e.code == StreamErrorCode.FLOW_CONTROL_ERROR: print("Flow control error on stream") elif e.code == StreamErrorCode.PROTOCOL_ERROR: print("Protocol error") ``` -------------------------------- ### Enable Compression Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/utilities.md Configure transport to enable gzip, brotli, or zstd compression. ```python # Keep defaults unless bandwidth is not a constraint transport = HTTPTransport( enable_gzip=True, enable_brotli=True, enable_zstd=True, ) ``` -------------------------------- ### Configure TLS and mTLS settings Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/configuration.md Shows how to load PEM-formatted certificates and keys for custom CA validation or client-side mTLS authentication. ```python # Custom CA certificate (e.g., for internal services) with open("ca.pem", "rb") as f: ca_cert = f.read() transport = HTTPTransport(tls_ca_cert=ca_cert) # mTLS with client certificate with open("client-key.pem", "rb") as f: key = f.read() with open("client-cert.pem", "rb") as f: cert = f.read() transport = HTTPTransport(tls_key=key, tls_cert=cert) ``` -------------------------------- ### Execute DELETE Request Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes a DELETE HTTP request. Parameters and behavior match get(). ```python async def delete( url: str, headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None = None, *, params: dict[str, str | None] | Iterable[tuple[str, str | None]] | None = None, ) -> FullResponse ``` -------------------------------- ### Configure Async HTTPX Client with Pyqwest Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Demonstrates wrapping a Pyqwest transport with retry logic for use in an asynchronous HTTPX client. ```python import asyncio import httpx from pyqwest import HTTPTransport, HTTPVersion from pyqwest.httpx import AsyncPyqwestTransport from pyqwest.middleware.retry import RetryTransport async def main(): # Configure pyqwest with HTTP/2, retries, TLS base_transport = HTTPTransport( http_version=HTTPVersion.HTTP2, timeout=30.0, tls_ca_cert=open("ca.pem", "rb").read(), ) # Add retry logic retry_transport = RetryTransport(base_transport, max_retries=3) # Wrap for httpx httpx_transport = AsyncPyqwestTransport(retry_transport) # Use with httpx async with httpx.AsyncClient(transport=httpx_transport) as client: response = await client.get("https://api.example.com/users") users = response.json() # POST with retry new_user = await client.post( "https://api.example.com/users", json={"name": "Alice", "email": "alice@example.com"} ) return users, new_user result = asyncio.run(main()) ``` -------------------------------- ### Async HTTP/2 Benchmark Results Source: https://github.com/curioswitch/pyqwest/blob/main/README.md Performance metrics for async HTTP/2 requests on a macOS laptop. ```text test_benchmark_async[pyqwest-0-http-h2] 23.8466 (1.0) 28.1299 (1.0) 25.3533 (1.0) 0.7858 (1.0) 25.2118 (1.0) 0.6631 (1.0) 5;3 39.4427 (1.0) 36 1 test_benchmark_async[httpx_pyqwest-0-http-h2] 60.2672 (2.53) 93.4816 (3.32) 63.5238 (2.51) 8.0387 (10.23) 61.4938 (2.44) 1.5876 (2.39) 1;1 15.7421 (0.40) 16 1 test_benchmark_async[httpx-0-http-h2] 180.1868 (7.56) 195.9454 (6.97) 184.1702 (7.26) 6.2000 (7.89) 181.3329 (7.19) 5.8279 (8.79) 1;1 5.4298 (0.14) 6 1 ``` -------------------------------- ### Import MeterProvider Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/types.md Import the MeterProvider class from the opentelemetry metrics module. ```python from opentelemetry.metrics import MeterProvider ``` -------------------------------- ### Retrieve header values Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/headers.md Access header values using dictionary-style indexing or the get method. ```python value = headers["Content-Type"] value = headers[HTTPHeaderName.CONTENT_TYPE] # Case-insensitive ``` ```python value = headers.get("X-Custom-Header", "default") ``` -------------------------------- ### Initialize SyncHTTPTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/transport.md Constructor signature for creating a new synchronous HTTP transport instance. ```python SyncHTTPTransport( *, tls_ca_cert: bytes | None = None, tls_key: bytes | None = None, tls_cert: bytes | None = None, http_version: HTTPVersion | None = None, timeout: float | None = None, connect_timeout: float | None = 30.0, read_timeout: float | None = None, pool_idle_timeout: float | None = 90.0, pool_max_idle_per_host: int | None = None, tcp_keepalive_interval: float | None = 30.0, enable_gzip: bool = True, enable_brotli: bool = True, enable_zstd: bool = True, use_system_dns: bool = False, enable_cookie_store: bool = False, enable_otel: bool = True, meter_provider: MeterProvider | None = None, tracer_provider: TracerProvider | None = None, ) -> SyncHTTPTransport ``` -------------------------------- ### Project Module Structure Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md The directory layout of the pyqwest package, highlighting core modules and middleware. ```text pyqwest/ ├── __init__.py # Main exports ├── _pyqwest.pyi # Type stubs for Rust bindings ├── _coro.py # Async Client wrapper ├── _errors.py # StreamError, StreamErrorCode ├── middleware/ │ └── retry/ │ ├── __init__.py │ ├── _async.py # RetryTransport (async) │ ├── _sync.py # SyncRetryTransport (sync) │ └── _shared.py # Shared retry logic └── httpx/ ├── __init__.py └── _transport.py # HTTPX adapter ``` -------------------------------- ### Manage HTTP Headers Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md Demonstrates creating a Headers object and adding both single and multi-value headers using case-insensitive keys. ```python from pyqwest import Headers, HTTPHeaderName headers = Headers() headers[HTTPHeaderName.CONTENT_TYPE] = "application/json" headers.add("Set-Cookie", "session=abc") # Multi-value headers.add("Set-Cookie", "path=/") # Multiple values ``` -------------------------------- ### FullResponse Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Initializes a new fully buffered response object. ```python FullResponse( status: int, headers: Headers, content: bytes, trailers: Headers, ) -> FullResponse ``` -------------------------------- ### SyncRequest Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Initializes a new synchronous HTTP request with method, URL, headers, content, and query parameters. ```python SyncRequest( method: str, url: str, headers: Headers | None = None, content: bytes | Iterable[bytes] | Mapping[str, Any] | None = None, *, params: dict[str, str | None] | Iterable[tuple[str, str | None]] | None = None, ) -> SyncRequest ``` -------------------------------- ### Configure HTTP/2 with AsyncPyqwestTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Enables HTTP/2 support by wrapping an HTTPTransport with the AsyncPyqwestTransport. ```python async with httpx.AsyncClient( transport=AsyncPyqwestTransport(HTTPTransport( http_version=HTTPVersion.HTTP2 )) ) as client: # Connection automatically upgraded to HTTP/2 response = await client.get("https://http2.example.com") print(response.http_version) # "HTTP/2" ``` -------------------------------- ### SyncResponse Context Manager Usage Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Demonstrates using the response as a context manager to iterate over streaming content. ```python with response: for chunk in response.content: # Process chunk pass ``` ```python def process_response(): response = transport.execute_sync(request) # Process streaming content with response: for chunk in response.content: print(len(chunk)) ``` -------------------------------- ### Initialize AsyncPyqwestTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Wraps a pyqwest HTTPTransport instance for use with an httpx.AsyncClient. ```python import httpx from pyqwest import HTTPTransport from pyqwest.httpx import AsyncPyqwestTransport # Create pyqwest transport pyqwest_transport = HTTPTransport(timeout=30.0) # Wrap for httpx httpx_transport = AsyncPyqwestTransport(pyqwest_transport) # Use with httpx.AsyncClient async with httpx.AsyncClient(transport=httpx_transport) as client: response = await client.get("https://httpbin.org/get") print(response.status_code) ``` -------------------------------- ### Create a custom timeout wrapper for async requests Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/utilities.md Uses asyncio.wait_for to enforce a timeout on client GET requests. Raises asyncio.TimeoutError if the operation exceeds the specified duration. ```python import asyncio from pyqwest import Client async def get_with_timeout(client: Client, url: str, timeout_secs: float): """Execute GET request with timeout.""" try: return await asyncio.wait_for( client.get(url), timeout=timeout_secs ) except asyncio.TimeoutError: print(f"Request to {url} timed out after {timeout_secs}s") raise async def main(): client = Client() try: response = await get_with_timeout( client, "https://slow-api.example.com", timeout_secs=5.0 ) print(f"Status: {response.status}") except asyncio.TimeoutError: pass asyncio.run(main()) ``` -------------------------------- ### Use PyqwestTransport with httpx.Client Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Demonstrates wrapping a pyqwest SyncHTTPTransport for use with an httpx.Client instance. ```python import httpx from pyqwest import SyncHTTPTransport from pyqwest.httpx import PyqwestTransport # Create pyqwest sync transport pyqwest_transport = SyncHTTPTransport(timeout=10.0) # Wrap for httpx httpx_transport = PyqwestTransport(pyqwest_transport) # Use with httpx.Client with httpx.Client(transport=httpx_transport) as client: response = client.get("https://httpbin.org/get") print(response.status_code) ``` -------------------------------- ### Configure Custom Transport with TLS Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md Initializes a client with a custom HTTPTransport including TLS certificate configuration. ```python from pyqwest import HTTPTransport, Client transport = HTTPTransport( tls_ca_cert=open("ca.pem", "rb").read(), timeout=30.0, ) client = Client(transport=transport) response = await client.get("https://internal-api.example.com") ``` -------------------------------- ### Initialize PyqwestTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Defines the constructor signature for PyqwestTransport. ```python PyqwestTransport(transport: SyncHTTPTransport) -> PyqwestTransport ``` -------------------------------- ### Async HTTP/1 Benchmark Results Source: https://github.com/curioswitch/pyqwest/blob/main/README.md Performance metrics for async HTTP/1 requests, including AIOHTTP comparisons. ```text test_benchmark_async[aiohttp-0-http-h1] 16.7938 (1.0) 20.7214 (1.0) 18.1937 (1.0) 0.7742 (1.86) 18.1273 (1.0) 1.0262 (1.90) 10;1 54.9642 (1.0) 38 1 test_benchmark_async[pyqwest-0-http-h1] 20.2705 (1.21) 22.2955 (1.08) 21.2425 (1.17) 0.4158 (1.0) 21.2635 (1.17) 0.5392 (1.0) 6;1 47.0754 (0.86) 31 1 test_benchmark_async[httpx_pyqwest-0-http-h1] 54.8534 (3.27) 88.8888 (4.29) 60.6263 (3.33) 9.7253 (23.39) 56.7566 (3.13) 2.4687 (4.58) 2;3 16.4945 (0.30) 18 1 test_benchmark_async[httpx-0-http-h1] 308.5213 (18.37) 333.9165 (16.11) 320.0622 (17.59) 11.2062 (26.95) 317.3413 (17.51) 20.1658 (37.40) 2;0 3.1244 (0.06) 5 1 ``` -------------------------------- ### options(url, headers, params) Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes an OPTIONS HTTP request. ```APIDOC ## options(url, headers, params) ### Description Executes an OPTIONS HTTP request. ### Parameters - **url** (str) - Required - **headers** (Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None) - Optional - **params** (dict[str, str | None] | Iterable[tuple[str, str | None]] | None) - Optional ### Returns - **FullResponse** ``` -------------------------------- ### Load TLS Certificates from Files Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/configuration.md Configure HTTPTransport with custom CA certificates or client certificate/key pairs read from the filesystem. ```python # Custom CA certificate with open("certs/ca.pem", "rb") as f: ca_cert = f.read() transport = HTTPTransport(tls_ca_cert=ca_cert) # Client certificate and key with open("certs/client.pem", "rb") as f: client_cert = f.read() with open("certs/client-key.pem", "rb") as f: client_key = f.read() transport = HTTPTransport(tls_cert=client_cert, tls_key=client_key) ``` -------------------------------- ### Request Constructor Definition Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md The signature for initializing a new async HTTP request. ```python Request( method: str, url: str, headers: Headers | None = None, content: bytes | AsyncIterator[bytes] | Mapping[str, Any] | None = None, *, params: dict[str, str | None] | Iterable[tuple[str, str | None]] | None = None, ) -> Request ``` -------------------------------- ### Configure MeterProvider in HTTPTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/types.md Initialize a MeterProvider and pass it to the HTTPTransport constructor for custom metrics collection. ```python from opentelemetry.sdk.metrics import MeterProvider from pyqwest import HTTPTransport meter_provider = MeterProvider() transport = HTTPTransport(meter_provider=meter_provider) ``` -------------------------------- ### Usage of default transport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/transport.md Demonstrates importing and retrieving the shared default transport instance. ```python from pyqwest import get_default_transport transport = get_default_transport() # This is reused by all Client instances with transport=None ``` -------------------------------- ### Configure HTTP Headers with HTTPHeaderName Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/configuration.md Use predefined constants for type-safe header management and case-insensitive matching. ```python from pyqwest import HTTPHeaderName, Headers headers = Headers() headers[HTTPHeaderName.CONTENT_TYPE] = "application/json" headers[HTTPHeaderName.AUTHORIZATION] = "Bearer token" headers.add(HTTPHeaderName.ACCEPT, "application/json") headers.add(HTTPHeaderName.ACCEPT, "text/plain") ``` -------------------------------- ### Use HTTPHeaderName for type-safe header access Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/headers.md Demonstrates setting and retrieving headers using HTTPHeaderName constants, highlighting case-insensitive access and type-safe checks. ```python from pyqwest import Headers, HTTPHeaderName headers = Headers() headers[HTTPHeaderName.CONTENT_TYPE] = "application/json" headers.add(HTTPHeaderName.ACCEPT, "text/plain") # Case-insensitive access print(headers[HTTPHeaderName.CONTENT_TYPE]) # "application/json" print(headers["content-type"]) # "application/json" # Type-safe operations if HTTPHeaderName.AUTHORIZATION not in headers: headers[HTTPHeaderName.AUTHORIZATION] = "Bearer token" ``` -------------------------------- ### Configure Query Parameters Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/configuration.md Pass query parameters using dictionaries, iterables, or key-only syntax. ```python from pyqwest import Client client = Client() # Dictionary syntax response = await client.get( "https://example.com/search", params={"q": "python", "sort": "relevance"} ) # Iterable syntax response = await client.get( "https://example.com/search", params=[("q", "python"), ("sort", "relevance")] ) # With None values (key-only parameters) response = await client.get( "https://example.com/api", params={"required": "value", "flag": None} ) # Results in: ?required=value&flag ``` -------------------------------- ### Implement Request/Response Logging with Async Transport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/utilities.md Wraps an existing transport to print request and response details to the console. Requires an asynchronous environment to execute. ```python from pyqwest import Client, HTTPTransport, Transport, Request, Response class LoggingTransport: """Transport wrapper that logs requests and responses.""" def __init__(self, transport: Transport): self._transport = transport async def execute(self, request: Request) -> Response: print(f">>> {request.method} {request.url}") for name, value in request.headers.items(): print(f" {name}: {value}") try: response = await self._transport.execute(request) print(f"<<< {response.status} ({response.http_version})") for name, value in response.headers.items(): print(f" {name}: {value}") return response except Exception as e: print(f"!!! {type(e).__name__}: {e}") raise async def main(): base = HTTPTransport(timeout=10.0) logged = LoggingTransport(base) client = Client(transport=logged) response = await client.get("https://httpbin.org/get") print(f"\nBody: {response.text()[:100]}...") asyncio.run(main()) ``` -------------------------------- ### SyncRequest Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Initializes a new synchronous HTTP request object. ```APIDOC ## SyncRequest Constructor ### Description Creates a new synchronous HTTP request. ### Signature SyncRequest(method: str, url: str, headers: Headers | None = None, content: bytes | Iterable[bytes] | Mapping[str, Any] | None = None, *, params: dict[str, str | None] | Iterable[tuple[str, str | None]] | None = None) -> SyncRequest ### Parameters - **method** (str) - Required - HTTP method. - **url** (str) - Required - Unencoded request URL. - **headers** (Headers | None) - Optional - Request headers. - **content** (bytes | Iterable[bytes] | Mapping[str, Any] | None) - Optional - Request body. Dictionary is JSON-encoded. Iterable allows streaming. - **params** (dict[str, str | None] | Iterable[tuple[str, str | None]] | None) - Optional - Query parameters. ``` -------------------------------- ### AsyncPyqwestTransport Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Initializes a new instance of AsyncPyqwestTransport by wrapping an existing pyqwest HTTPTransport instance. ```APIDOC ## AsyncPyqwestTransport(transport: HTTPTransport) ### Description Creates an async transport for use with httpx.AsyncClient, delegating requests to the provided pyqwest HTTPTransport. ### Parameters - **transport** (HTTPTransport) - Required - The pyqwest HTTPTransport to delegate requests to. ### Example ```python import httpx from pyqwest import HTTPTransport from pyqwest.httpx import AsyncPyqwestTransport # Create pyqwest transport pyqwest_transport = HTTPTransport(timeout=30.0) # Wrap for httpx httpx_transport = AsyncPyqwestTransport(pyqwest_transport) # Use with httpx.AsyncClient async with httpx.AsyncClient(transport=httpx_transport) as client: response = await client.get("https://httpbin.org/get") print(response.status_code) ``` ``` -------------------------------- ### execute(method, url, headers, content, params) Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes an HTTP request with an arbitrary method, returning the full buffered response. ```APIDOC ## execute(method, url, headers, content, params) ### Description Executes an HTTP request with an arbitrary method, returning the full buffered response. ### Parameters - **method** (str) - Required - The HTTP method (e.g., "GET", "POST", "CUSTOM"). - **url** (str) - Required - The unencoded request URL. - **headers** (Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None) - Optional - The request headers. - **content** (bytes | AsyncIterator[bytes] | Mapping[str, Any] | None) - Optional - The request content. - **params** (dict[str, str | None] | Iterable[tuple[str, str | None]] | None) - Optional - Query parameters. ### Returns - **FullResponse** - The fully buffered HTTP response. ``` -------------------------------- ### Import TracerProvider Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/types.md Import the TracerProvider class from the opentelemetry trace module. ```python from opentelemetry.trace import TracerProvider ``` -------------------------------- ### Sync HTTP/2 Benchmark Results Source: https://github.com/curioswitch/pyqwest/blob/main/README.md Performance metrics for synchronous HTTP/2 requests. ```text test_benchmark_sync[pyqwest-0-http-h2] 14.3617 (1.0) 16.8924 (1.0) 14.9348 (1.0) 0.4427 (1.0) 14.8650 (1.0) 0.4911 (1.0) 11;2 66.9576 (1.0) 52 1 test_benchmark_sync[httpx_pyqwest-0-http-h2] 52.0353 (3.62) 100.9097 (5.97) 56.7675 (3.80) 12.0905 (27.31) 52.6683 (3.54) 0.8969 (1.83) 2;4 17.6157 (0.26) 19 1 test_benchmark_sync[httpx-0-http-h2] 97.2703 (6.77) 131.0486 (7.76) 101.3821 (6.79) 10.4656 (23.64) 97.8609 (6.58) 0.8371 (1.70) 1;2 9.8637 (0.15) 10 1 ``` -------------------------------- ### Initialize Headers object Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/headers.md Construct a new Headers instance using a dictionary, an iterable of tuples, or HTTPHeaderName constants. ```python from pyqwest import Headers, HTTPHeaderName # Empty headers headers = Headers() # From dictionary headers = Headers({"Content-Type": "application/json", "Authorization": "Bearer token"}) # From iterable of tuples headers = Headers([("Content-Type", "application/json"), ("Accept", "text/plain")]) # Using HTTPHeaderName constants headers = Headers({ HTTPHeaderName.CONTENT_TYPE: "application/json", HTTPHeaderName.ACCEPT: "text/plain", }) ``` -------------------------------- ### Construct an HTTPHeaderName object Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/headers.md Instantiate a header name object. Use class attributes for standard headers instead of manual instantiation. ```python HTTPHeaderName(name: str) -> HTTPHeaderName ``` -------------------------------- ### HTTP Client Methods Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md Standard HTTP methods supported by both sync and async clients. ```APIDOC ## HTTP Client Methods ### Description Standard HTTP methods available on both synchronous and asynchronous clients. ### Methods - **GET**: `client.get()` - **POST**: `client.post()` - **PUT**: `client.put()` - **PATCH**: `client.patch()` - **DELETE**: `client.delete()` - **HEAD**: `client.head()` - **OPTIONS**: `client.options()` ``` -------------------------------- ### Initialize SyncRetryTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/retry-middleware.md Constructor signature for configuring synchronous retry behavior. ```python SyncRetryTransport( transport: SyncTransport, initial_interval: float = 0.5, randomization_factor: float = 0.5, multiplier: float = 1.5, max_interval: float = 60.0, max_retries: int = 4, ) -> SyncRetryTransport ``` -------------------------------- ### Configure TLS with CA Certificate Source: https://github.com/curioswitch/pyqwest/blob/main/docs/usage.md Initialize a transport with a custom CA certificate for secure connections. ```python import asyncio from pathlib import Path from pyqwest import Client, HTTPTransport ca_cert = asyncio.to_thread(Path("/certs/ca.crt").read) async with HTTPTransport(tls_ca_cert=ca_cert) as transport: client = Client(transport) application = MyApplication(client) ``` ```python from pathlib import Path from pyqwest import SyncClient, SyncHTTPTransport ca_cert = Path("/certs/ca.crt").read() with SyncHTTPTransport(tls_ca_cert=my_cert) as transport: client = SyncClient(transport) application = MyApplication(client) ``` -------------------------------- ### Migrate httpx client to pyqwest transport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Replace the default httpx transport by initializing AsyncPyqwestTransport and passing it to the AsyncClient constructor. ```python # Before: Using httpx's default HTTP client import httpx client = httpx.AsyncClient() # After: Using pyqwest via httpx from pyqwest import HTTPTransport from pyqwest.httpx import AsyncPyqwestTransport transport = AsyncPyqwestTransport(HTTPTransport()) client = httpx.AsyncClient(transport=transport) # API remains identical; underlying implementation is pyqwest ``` -------------------------------- ### SyncResponse Constructor Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Initializes a new streaming synchronous response object. ```python SyncResponse( *, status: int, http_version: HTTPVersion | None = None, headers: Headers | None = None, content: bytes | Iterable[bytes] | None = None, trailers: Headers | None = None, ) -> SyncResponse ``` -------------------------------- ### Initialize Response Object Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/request-response.md Constructor for creating a new streaming async response. Typically used internally by transport.execute(). ```python Response( *, status: int, http_version: HTTPVersion | None = None, headers: Headers | None = None, content: bytes | AsyncIterator[bytes] | None = None, trailers: Headers | None = None, ) -> Response ``` -------------------------------- ### Reuse Client Instances Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/utilities.md Avoid creating a new client for every request to improve performance. ```python # Good client = Client() for url in urls: response = await client.get(url) # Bad for url in urls: client = Client() # Creates new client each time response = await client.get(url) ``` -------------------------------- ### Handle Pyqwest Errors in HTTPX Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Demonstrates catching standard HTTPX exceptions that map to underlying Pyqwest stream and protocol errors. ```python import asyncio import httpx from pyqwest import HTTPTransport, StreamError, StreamErrorCode from pyqwest.httpx import AsyncPyqwestTransport async def main(): httpx_transport = AsyncPyqwestTransport(HTTPTransport()) async with httpx.AsyncClient(transport=httpx_transport) as client: try: response = await client.get("https://api.example.com") except httpx.RemoteProtocolError as e: # Raised from StreamError in pyqwest print(f"HTTP/2 protocol error: {e}") except httpx.ConnectError as e: # Network error print(f"Connection failed: {e}") except httpx.TimeoutException as e: # Timeout print(f"Request timed out: {e}") asyncio.run(main()) ``` -------------------------------- ### AsyncPyqwestTransport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Initializes an httpx AsyncClient with the AsyncPyqwestTransport to enable advanced protocol support. ```APIDOC ## AsyncPyqwestTransport ### Description Integrates pyqwest transport into an httpx.AsyncClient to enable features such as HTTP/2 and HTTP/3 support. ### Usage ```python async with httpx.AsyncClient( transport=AsyncPyqwestTransport(HTTPTransport( http_version=HTTPVersion.HTTP2 )) ) as client: response = await client.get("https://http2.example.com") ``` ``` -------------------------------- ### Compare HTTPHeaderName objects Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/headers.md HTTPHeaderName instances support standard equality and inequality comparison operators. ```python def __eq__(self, other: object) -> bool def __ne__(self, other: object) -> bool ``` -------------------------------- ### Configure HTTP Transport Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/utilities.md Configure transport settings once during client initialization. ```python transport = HTTPTransport(timeout=30.0, pool_max_idle_per_host=10) client = Client(transport=transport) ``` -------------------------------- ### head(url, headers, params) Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/client.md Executes a HEAD HTTP request. ```APIDOC ## head(url, headers, params) ### Description Executes a HEAD HTTP request. ### Parameters - **url** (str) - Required - **headers** (Headers | Mapping[str, str] | Iterable[tuple[str, str]] | None) - Optional - **params** (dict[str, str | None] | Iterable[tuple[str, str | None]] | None) - Optional ### Returns - **FullResponse** ``` -------------------------------- ### Integrate with HTTPX Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/OVERVIEW.md Uses Pyqwest transport within an HTTPX AsyncClient instance. ```python import httpx from pyqwest import HTTPTransport from pyqwest.httpx import AsyncPyqwestTransport transport = AsyncPyqwestTransport(HTTPTransport()) async with httpx.AsyncClient(transport=transport) as client: response = await client.get("https://httpbin.org/get") ``` -------------------------------- ### Map HTTPX Timeouts to Pyqwest Source: https://github.com/curioswitch/pyqwest/blob/main/_autodocs/api-reference/httpx-adapter.md Demonstrates how HTTPX timeout objects are automatically converted to Pyqwest timeout semantics. ```python import httpx # httpx timeout format timeout = httpx.Timeout( connect=5.0, read=10.0, write=10.0, pool=5.0 ) async with httpx.AsyncClient( transport=AsyncPyqwestTransport(HTTPTransport()), timeout=timeout ) as client: # Mapped to pyqwest timeout semantics response = await client.get("https://example.com") ```