### Trio Example with httpx2 Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Shows how to use httpx2 for asynchronous HTTP requests within the Trio framework. The 'trio' package must be installed separately. ```python import httpx2 import trio async def main(): async with httpx2.AsyncClient() as client: response = await client.get('https://www.example.com/') print(response) trio.run(main) ``` -------------------------------- ### Install HTTPX2 Source: https://github.com/pydantic/httpx2/blob/main/README.md Install the httpx2 library using pip. This is the basic installation for the HTTP client. ```shell pip install httpx2 ``` -------------------------------- ### Install Project Dependencies Source: https://github.com/pydantic/httpx2/blob/main/docs/contributing.md Navigate to the project directory and install the project and its dependencies using the provided script. ```shell cd httpx2 scripts/install ``` -------------------------------- ### Install HTTP Core Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/README.md Install the base httpcore package for HTTP/1.1 only support. ```shell pip install httpcore ``` -------------------------------- ### Install HTTP Core with Optional Extras Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/README.md Install httpcore with optional extras like asyncio, trio, http2, and socks support. ```shell pip install httpcore['asyncio,trio,http2,socks'] ``` -------------------------------- ### AsyncIO Example with httpx2 Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Demonstrates making an asynchronous HTTP GET request using httpx2 with the built-in AsyncIO library. Ensure AsyncIO is available in your Python environment. ```python import asyncio import httpx2 async def main(): async with httpx2.AsyncClient() as client: response = await client.get('https://www.example.com/') print(response) asyncio.run(main()) ``` -------------------------------- ### Install Asyncio Support Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Install the optional dependencies for using HTTPX with Python's stdlib asyncio. ```shell pip install 'httpcore[asyncio]' ``` -------------------------------- ### Install HTTPCore with HTTP/2 Support Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/index.md Install HTTPCore with the extra dependency for HTTP/1.1 and HTTP/2 support. ```shell pip install httpcore[http2] ``` -------------------------------- ### Install Trio Support Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Install the optional dependencies for using HTTPX with the Python trio package. ```shell pip install 'httpcore[trio]' ``` -------------------------------- ### Import HTTPX2 Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Start by importing the httpx2 library. ```python import httpx2 ``` -------------------------------- ### Install HTTPCore with SOCKS Proxy Support Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/index.md Install HTTPCore with the extra dependency for SOCKS proxy support. ```shell pip install httpcore[socks] ``` -------------------------------- ### Install HTTP/2 Dependencies Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/http2.md Install the optional HTTP/2 dependencies for httpcore using pip. ```shell pip install 'httpcore[http2]' ``` -------------------------------- ### Making an Async GET Request Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Demonstrates how to create an AsyncClient and perform a GET request within an async context manager. ```pycon >>> async with httpx2.AsyncClient() as client: ... r = await client.get('https://www.example.com/') ... >>> r ``` -------------------------------- ### Installing chardet for Auto-Detection Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/text-encodings.md Installs the necessary packages for enabling character-set auto-detection with httpx2. ```shell pip install httpx2 pip install chardet ``` -------------------------------- ### Install httpx2 with CLI support Source: https://github.com/pydantic/httpx2/blob/main/README.md Install httpx2 with the optional command-line client dependency. This enables using httpx2 directly from the terminal. ```shell pip install 'httpx2[cli]' ``` -------------------------------- ### Install httpx2 with HTTP/2 support Source: https://github.com/pydantic/httpx2/blob/main/docs/http2.md Install the optional HTTP/2 dependencies for httpx2 using pip. ```shell pip install 'httpx2[http2]' ``` -------------------------------- ### Install httpx2 with Brotli and Zstandard support Source: https://github.com/pydantic/httpx2/blob/main/docs/index.md Install httpx2 with optional brotli and zstandard decoders support. ```shell pip install 'httpx2[brotli,zstd]' ``` -------------------------------- ### Install httpx2 with HTTP/2 support Source: https://github.com/pydantic/httpx2/blob/main/README.md Install httpx2 with the optional HTTP/2 protocol support. This is required for making HTTP/2 requests. ```shell pip install httpx2[http2] ``` -------------------------------- ### AnyIO Example with httpx2 (Trio Backend) Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Illustrates using httpx2 with AnyIO, specifying the 'trio' backend. AnyIO provides a unified API for different async libraries. ```python import httpx2 import anyio async def main(): async with httpx2.AsyncClient() as client: response = await client.get('https://www.example.com/') print(response) anyio.run(main, backend='trio') ``` -------------------------------- ### Configuring Request Timeouts Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/extensions.md This example shows how to configure specific timeout values for connection establishment and pool waiting using the 'timeout' extension. ```python # Timeout if a connection takes more than 5 seconds to established, or if # we are blocked waiting on the connection pool for more than 10 seconds. r = httpcore.request( "GET", "https://www.example.com", extensions={\"timeout\": {\"connect\": 5.0, \"pool\": 10.0}} ) ``` -------------------------------- ### Handling Websockets with wsproto Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/extensions.md Demonstrates how to use the wsproto package with httpcore to establish and manage a websockets session. Requires wsproto to be installed. ```python import httpcore import wsproto import os import base64 url = "http://127.0.0.1:8000/" headers = { b"Connection": b"Upgrade", b"Upgrade": b"WebSocket", b"Sec-WebSocket-Key": base64.b64encode(os.urandom(16)), b"Sec-WebSocket-Version": b"13" } with httpcore.stream("GET", url, headers=headers) as response: if response.status != 101: raise Exception("Failed to upgrade to websockets", response) # Get the raw network stream. network_steam = response.extensions["network_stream"] # Write a WebSocket text frame to the stream. ws_connection = wsproto.Connection(wsproto.ConnectionType.CLIENT) message = wsproto.events.TextMessage("hello, world!") outgoing_data = ws_connection.send(message) network_steam.write(outgoing_data) # Wait for a response. incoming_data = network_steam.read(max_bytes=4096) ws_connection.receive_data(incoming_data) for event in ws_connection.events(): if isinstance(event, wsproto.events.TextMessage): print("Got data:", event.data) # Write a WebSocket close to the stream. message = wsproto.events.CloseConnection(code=1000) outgoing_data = ws_connection.send(message) network_steam.write(outgoing_data) ``` -------------------------------- ### Instantiate and Use a Connection Pool Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/connection-pools.md Demonstrates the basic usage of httpcore's ConnectionPool to send a GET request and print the response. ```python import httpcore http = httpcore.ConnectionPool() r = http.request("GET", "https://www.example.com/") print(r) # ``` -------------------------------- ### Install SOCKS Proxy Support Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/proxies.md Install the optional SOCKS support for httpx2 using pip. This is required before configuring SOCKS proxies. ```shell pip install 'httpx2[socks]' ``` -------------------------------- ### Make a GET Request Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Make a simple GET request to a URL and check the response status. ```python r = httpx2.get('https://httpbin.org/get') r ``` -------------------------------- ### Make a GET Request with Custom Headers Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md Demonstrates sending a request with custom headers using a Client instance. ```python >>> with httpx2.Client() as client: ... headers = {'X-Custom': 'value'} ... r = client.get('https://example.com', headers=headers) ... >>> r.request.headers['X-Custom'] 'value' ``` -------------------------------- ### Basic GET Request with httpx2 Source: https://github.com/pydantic/httpx2/blob/main/README.md Demonstrates making a simple GET request to a URL and accessing response details like status code, headers, and text content. ```python >>> import httpx2 >>> r = httpx2.get('https://www.example.org/') >>> r >>> r.status_code 200 >>> r.headers['content-type'] 'text/html; charset=UTF-8' >>> r.text '\n\n\nExample Domain...' ``` -------------------------------- ### Basic Authentication Example Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/authentication.md Demonstrates using Basic Authentication with a client instance for a specific HTTP endpoint. This is suitable for unencrypted connections over HTTPS. ```python >>> auth = httpx2.BasicAuth(username="finley", password="secret") >>> client = httpx2.Client(auth=auth) >>> response = client.get("https://httpbin.org/basic-auth/finley/secret") >>> response ``` -------------------------------- ### Create and Inspect an HTTP Request Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/requests-responses-urls.md Demonstrates how to create a basic HTTP GET request and inspect its method, URL, headers, and stream. The URL is represented as a structured httpcore.URL object. ```python >>> request = httpcore.Request("GET", "https://www.example.com/") >>> request.method b"GET" >>> request.url httpcore.URL(scheme=b"https", host=b"www.example.com", port=None, target=b"/") >>> request.headers [(b'Host', b'www.example.com')] >>> request.stream ``` -------------------------------- ### Create Image from Binary Content Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Example of creating an image from binary response content using Pillow and BytesIO. ```python from PIL import Image from io import BytesIO i = Image.open(BytesIO(r.content)) ``` -------------------------------- ### Sending Async Requests Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Shows how to send a GET request asynchronously using httpcore. Requires the 'await' keyword. ```python import asyncio import httpcore async def main(): async with httpcore.AsyncConnectionPool() as http: response = await http.request("GET", "https://www.example.com/") asyncio.run(main()) ``` -------------------------------- ### Make a GET Request with Client Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md Send requests using client methods like .get(). These methods accept the same arguments as the top-level httpx2 functions. ```python >>> with httpx2.Client() as client: ... r = client.get('https://example.com') ... >>> r ``` -------------------------------- ### Using the Trace Extension for Event Monitoring Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/extensions.md This snippet demonstrates how to use the 'trace' extension to install a callback handler for monitoring internal httpcore events. The callback logs event names and associated information. ```python import httpcore def log(event_name, info): print(event_name, info) r = httpcore.request("GET", "https://www.example.com/", extensions={\"trace\": log}) # connection.connect_tcp.started {'host': 'www.example.com', 'port': 443, 'local_address': None, 'timeout': None} # connection.connect_tcp.complete {'return_value': } # connection.start_tls.started {'ssl_context': , 'server_hostname': b'www.example.com', 'timeout': None} # connection.start_tls.complete {'return_value': } # http11.send_request_headers.started {'request': } # http11.send_request_headers.complete {'return_value': None} # http11.send_request_body.started {'request': } # http11.send_request_body.complete {'return_value': None} # http11.receive_response_headers.started {'request': } # http11.receive_response_headers.complete {'return_value': (b'HTTP/1.1', 200, b'OK', [(b'Age', b'553715'), (b'Cache-Control', b'max-age=604800'), (b'Content-Type', b'text/html; charset=UTF-8'), (b'Date', b'Thu, 21 Oct 2021 17:08:42 GMT'), (b'Etag', b'"3147526947+ident"'), (b'Expires', b'Thu, 28 Oct 2021 17:08:42 GMT'), (b'Last-Modified', b'Thu, 17 Oct 2019 07:18:26 GMT'), (b'Server', b'ECS (nyb/1DCD)'), (b'Vary', b'Accept-Encoding'), (b'X-Cache', b'HIT'), (b'Content-Length', b'1256')])} # http11.receive_response_body.started {'request': } # http11.receive_response_body.complete {'return_value': None} # http11.response_closed.started {} ``` -------------------------------- ### Monitor Upload Progress with tqdm Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md Stream data for an upload request and display progress using the tqdm library. This example uses a generator to yield data chunks for the upload. ```python import io import random import httpx2 from tqdm import tqdm def gen(): """ this is a complete example with generated random bytes. you can replace `io.BytesIO` with real file object. """ total = 32 * 1024 * 1024 # 32m with tqdm(ascii=True, unit_scale=True, unit='B', unit_divisor=1024, total=total) as bar: with io.BytesIO(random.randbytes(total)) as f: while data := f.read(1024): yield data bar.update(len(data)) httpx2.post("https://httpbin.org/post", content=gen()) ``` -------------------------------- ### Dictionary-Based Logging Configuration Source: https://github.com/pydantic/httpx2/blob/main/docs/logging.md Use dictionary configuration for more complex logging setups, directing output to stderr. Ensure httpx2 is imported. ```python import logging.config import httpx2 LOGGING_CONFIG = { "version": 1, "handlers": { "default": { "class": "logging.StreamHandler", "formatter": "http", "stream": "ext://sys.stderr" } }, "formatters": { "http": { "format": "% (levelname)s [%(asctime)s] %(name)s - %(message)s", "datefmt": "%Y-%m-%d %H:%M:%S", } }, 'loggers': { 'httpx2': { 'handlers': ['default'], 'level': 'DEBUG', }, 'httpcore': { 'handlers': ['default'], 'level': 'DEBUG', }, } } logging.config.dictConfig(LOGGING_CONFIG) httpx2.get('https://www.example.com') ``` -------------------------------- ### Configure and Enable httpcore Logging Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/logging.md Use this snippet to configure Python's logging to output debug information from httpcore. Ensure you have both the `logging` and `httpcore` libraries installed. ```python import logging import httpcore logging.basicConfig( format="%(levelname)s [%(asctime)s] %(name)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.DEBUG ) httpcore.request('GET', 'https://www.example.com') ``` -------------------------------- ### Digest Authentication Example Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/authentication.md Demonstrates using Digest Authentication with a client instance. This scheme provides encryption and is suitable for unencrypted HTTP connections, requiring an initial challenge-response. ```python >>> auth = httpx2.DigestAuth(username="olivia", password="secret") >>> client = httpx2.Client(auth=auth) >>> response = client.get("https://httpbin.org/digest-auth/auth/olivia/secret") >>> response >>> response.history [] ``` -------------------------------- ### Simplified HTTPX2 Proxy Configuration Source: https://github.com/pydantic/httpx2/blob/main/docs/troubleshooting.md A more concise way to configure a proxy for all requests when the proxy supports HTTP connections. This simplifies the client setup. ```python proxy = "http://myproxy.org" with httpx2.Client(proxy=proxy) as client: ... ``` -------------------------------- ### Instantiate Client as Context Manager Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md The recommended way to use a Client is as a context manager to ensure proper connection cleanup. ```python with httpx2.Client() as client: ... ``` -------------------------------- ### Run Documentation Site Locally Source: https://github.com/pydantic/httpx2/blob/main/docs/contributing.md Build and serve the documentation site locally to preview changes. This is useful for documentation development. ```shell scripts/docs ``` -------------------------------- ### Instantiating AsyncClient with Context Manager Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Shows the recommended way to instantiate and use an AsyncClient using an 'async with' statement for automatic resource management. ```python async with httpx2.AsyncClient() as client: ... ``` -------------------------------- ### Configure Client with Default Truststore Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/ssl.md Instantiates an httpx2 client using the default operating system trust store for SSL verification. ```python import ssl import truststore import httpx2 # This SSL context is equivalent to the default `verify=True`. c_tx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT) client = httpx2.Client(verify=c_tx) ``` -------------------------------- ### SOCKS5 Proxy Configuration Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/proxies.md Configure httpcore to use a SOCKS5 proxy. Ensure the optional 'socks' dependency is installed (`pip install 'httpcore[socks]'`). ```python import httpcore # Note that the SOCKS port is 1080. proxy = httpcore.Proxy(url="socks5://127.0.0.1:1080/") pool = httpcore.ConnectionPool(proxy=proxy) r = pool.request("GET", "https://www.example.com/") ``` -------------------------------- ### Proxy All Requests to a Domain and its Subdomains Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/transports.md Uses a wildcard to route all requests to 'example.com' and any of its subdomains through a specified proxy. ```python mounts = { "all://*example.com": httpx2.HTTPTransport(proxy="http://localhost:8030"), } ``` -------------------------------- ### Get JSON Response Content Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Parse and access JSON content from an API response. ```python r = httpx2.get('https://api.github.com/events') r.json() ``` -------------------------------- ### Instantiate AsyncClient with HTTP/2 enabled Source: https://github.com/pydantic/httpx2/blob/main/docs/http2.md Enable HTTP/2 support when creating an httpx2.AsyncClient instance. ```python client = httpx2.AsyncClient(http2=True) ... ``` -------------------------------- ### Get Text Response Content Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Access the response content as decoded Unicode text. ```python r = httpx2.get('https://www.example.org/') r.text ``` -------------------------------- ### httpx2.get Source: https://github.com/pydantic/httpx2/blob/main/docs/api.md A helper function for making GET requests directly. Use this for simple, one-off requests. ```APIDOC ## httpx2.get ### Description Makes a GET HTTP request directly. Suitable for console testing or a small number of requests. ### Method GET ### Endpoint N/A (Function call) ### Parameters (Details not provided in source, refer to httpx2.Client.get for signature) ### Request Example (Not provided in source) ### Response (Details not provided in source, refer to httpx2.Response for details) ``` -------------------------------- ### Async Connection Pool Usage Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Demonstrates the basic usage of an async connection pool using a context manager. ```python # The async variation of `httpcore.ConnectionPool` async with httpcore.AsyncConnectionPool() as http: ... ``` -------------------------------- ### Get Binary Response Content Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Access the response content as raw bytes, suitable for non-text responses. ```python r.content ``` -------------------------------- ### Configure Client with Client-Side Certificates Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/ssl.md Sets up an httpx2 client to use client-side certificates for authentication, loading them using ssl.SSLContext.load_cert_chain. ```python ctx = ssl.create_default_context() ctx.load_cert_chain(certfile="path/to/client.pem") # Optionally also keyfile or password. client = httpx2.Client(verify=ctx) ``` -------------------------------- ### Pin HTTP Core Version Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/README.md Pin your httpcore requirements to a specific major version, such as '1.*', for stable installations. ```shell pip install 'httpcore==1.*' ``` -------------------------------- ### Pass URL Query Parameters Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Include URL query parameters in a GET request using the `params` keyword. ```python params = {'key1': 'value1', 'key2': 'value2'} r = httpx2.get('https://httpbin.org/get', params=params) ``` -------------------------------- ### Determining Next Redirect Request Source: https://github.com/pydantic/httpx2/blob/main/docs/compatibility.md HTTPX uses `response.next_request` to get the next redirect request, analogous to `requests.response.next`. ```python session = requests.Session() request = requests.Request("GET", ...).prepare() while request is not None: response = session.send(request, allow_redirects=False) request = response.next ``` ```python client = httpx2.Client() request = client.build_request("GET", ...) while request is not None: response = client.send(request) request = response.next_request ``` -------------------------------- ### Mock Requests to a Specific Domain Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/transports.md Sets up a mock transport to handle requests for 'example.org', returning a predefined JSON response. Other requests proceed normally. ```python # All requests to "example.org" should be mocked out. # Other requests occur as usual. def handler(request): return httpx2.Response(200, json={"text": "Hello, World!"}) mounts = {"all://example.org": httpx2.MockTransport(handler)} client = httpx2.Client(mounts=mounts) ``` -------------------------------- ### Creating Headers Source: https://github.com/pydantic/httpx2/blob/main/docs/api.md Illustrates the creation and usage of httpx2.Headers, a case-insensitive multi-dictionary for managing HTTP headers. It shows how to initialize headers and access values regardless of case. ```python >>> headers = Headers({'Content-Type': 'application/json'}) >>> headers['content-type'] 'application/json' ``` -------------------------------- ### Implement HTTPSRedirectTransport for AsyncClient Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/transports.md An example of a transport implementation for `AsyncClient` that redirects HTTP to HTTPS. Note the different signature for `handle_async_request`. ```python import httpx2 class HTTPSRedirectTransport(httpx2.BaseTransport): """ A transport that always redirects to HTTPS. """ def handle_request(self, method, url, headers, stream, extensions): scheme, host, port, path = url if port is None: location = b"https://%s%s" % (host, path) else: location = b"https://%s:%d%s" % (host, port, path) stream = httpx2.ByteStream(b"") headers = [(b"location", location)] extensions = {} return 303, headers, stream, extensions ``` -------------------------------- ### Download a Large File with Streaming Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/quickstart.md Demonstrates downloading a large binary file using `httpcore.stream` and writing it to a local file in chunks. ```python import httpcore with httpcore.stream('GET', 'https://speed.hetzner.de/100MB.bin') as response: with open("download.bin", "wb") as output_file: for chunk in response.iter_stream(): output_file.write(chunk) ``` -------------------------------- ### Configure Client with a SOCKS Proxy Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/proxies.md Set up a client to use a SOCKS proxy by providing the SOCKS proxy URL to the 'proxy' parameter. ```python httpx2.Client(proxy='socks5://user:pass@host:port') ``` -------------------------------- ### Access Response Headers Case-Insensitively Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Access response headers using any capitalization. The `get` method is also available for safe access. ```python r.headers['Content-Type'] ``` ```python r.headers.get('content-type') ``` -------------------------------- ### HTTP/1.1 Connection Pool Output Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/http2.md Example output from the HTTP/1.1 performance test, showing multiple connections used for parallel requests. ```text , , , , , , , Complete in 0.586 seconds ``` -------------------------------- ### Create a Basic Request Instance Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md Instantiate a basic HTTPX2 Request object with a specified method and URL. ```python request = httpx2.Request("GET", "https://example.com") ``` -------------------------------- ### Async HTTP Request with AnyIOBackend Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/network-backends.md Demonstrates making an asynchronous HTTP request using the AnyIOBackend, suitable for asyncio or trio environments. ```python import httpcore import asyncio async def main(): network_backend = httpcore.AnyIOBackend() async with httpcore.AsyncConnectionPool(network_backend=network_backend) as http: response = await http.request('GET', 'https://www.example.com') print(response) asyncio.run(main()) ``` -------------------------------- ### Stream a Response Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/quickstart.md Use `httpcore.stream` to handle large responses without reading the entire content into memory. Iterate over `response.iter_stream()` to get chunks. ```python import httpcore with httpcore.stream('GET', 'https://example.com') as response: for chunk in response.iter_stream(): print(f"Downloaded: {chunk}") ``` -------------------------------- ### Configure Client with Proxy Authentication Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/proxies.md Include username and password in the proxy URL using the 'userinfo' section to authenticate with the proxy server. ```python with httpx2.Client(proxy="http://username:password@localhost:8030") as client: ... ``` -------------------------------- ### Accessing Response URL in HTTPX Source: https://github.com/pydantic/httpx2/blob/main/docs/compatibility.md HTTPX's `response.url` returns a `URL` instance. Use `str(response.url)` to get a string representation. ```python Use `str(response.url)` if you need a string instance. ``` -------------------------------- ### Other HTTP Methods Source: https://github.com/pydantic/httpx2/blob/main/docs/quickstart.md Demonstrates making PUT, DELETE, HEAD, and OPTIONS requests. ```python r = httpx2.put('https://httpbin.org/put', data={'key': 'value'}) ``` ```python r = httpx2.delete('https://httpbin.org/delete') ``` ```python r = httpx2.head('https://httpbin.org/get') ``` ```python r = httpx2.options('https://httpbin.org/get') ``` -------------------------------- ### Using AsyncHTTPTransport with AsyncClient Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Demonstrates how to instantiate an AsyncClient with a custom AsyncHTTPTransport, specifying options like retries. ```pycon >>> import httpx2 >>> transport = httpx2.AsyncHTTPTransport(retries=1) >>> async with httpx2.AsyncClient(transport=transport) as client: >>> ... ``` -------------------------------- ### Multi-Request Authentication Flow Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/authentication.md Handle authentication flows that require multiple requests. This example shows how to resend a request with a custom header if the initial response is a 401. ```python class MyCustomAuth(httpx2.Auth): def __init__(self, token): self.token = token def auth_flow(self, request): response = yield request if response.status_code == 401: # If the server issues a 401 response then resend the request, # with a custom `X-Authentication` header. request.headers['X-Authentication'] = self.token yield request ``` -------------------------------- ### Configuring a Client with Proxy Source: https://github.com/pydantic/httpx2/blob/main/docs/api.md Shows how to configure an httpx2.Client to use a proxy server. This involves creating a Proxy object with the proxy's URL and passing it to the Client constructor. ```python >>> proxy = Proxy("http://proxy.example.com:8030") >>> client = Client(proxy=proxy) ``` -------------------------------- ### HTTP/2 Performance Test Output Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/http2.md Example output when running the concurrent requests script with an HTTP/2 enabled connection pool, showing all requests handled over a single connection. ```text Complete in 0.573 seconds ``` -------------------------------- ### View httpx2 CLI help Source: https://github.com/pydantic/httpx2/blob/main/README.md Displays the help message for the httpx2 command-line interface, showing available commands and options. ```shell httpx2 --help ``` -------------------------------- ### Basic Logging Configuration Source: https://github.com/pydantic/httpx2/blob/main/docs/logging.md Configure basic logging to output debug information from httpx2 and httpcore to stdout. Ensure httpx2 is imported. ```python import logging import httpx2 logging.basicConfig( format="%(levelname)s [%(asctime)s] %(name)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.DEBUG ) httpx2.get("https://www.example.com") ``` -------------------------------- ### Configure Client with Alternative Certificate Store Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/ssl.md Loads an alternative certificate verification store using the standard SSL context API for an httpx2 client. ```python import httpx2 import ssl # Use an explicitly configured certificate store. c_tx = ssl.create_default_context(cafile="path/to/certs.pem") # Either cafile or capath. client = httpx2.Client(verify=c_tx) ``` -------------------------------- ### Handling CONNECT Requests with Network Stream Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/extensions.md Demonstrates using the 'network_stream' extension to handle CONNECT requests, including upgrading to an SSL stream and manually sending/receiving data. ```python # Formulate a CONNECT request... # # This will establish a connection to 127.0.0.1:8080, and then send the following... # # CONNECT http://www.example.com HTTP/1.1 url = "http://127.0.0.1:8080" extensions = {"target: "http://www.example.com"} with httpcore.stream("CONNECT", url, extensions=extensions) as response: network_stream = response.extensions["network_stream"] # Upgrade to an SSL stream... network_stream = network_stream.start_tls( ssl_context=httpcore.default_ssl_context(), hostname=b"www.example.com", ) # Manually send an HTTP request over the network stream, and read the response... # # For a more complete example see the httpcore `TunnelHTTPConnection` implementation. network_stream.write(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n") data = network_stream.read() print(data) ``` -------------------------------- ### Create an HTTP Request with Custom Headers Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/requests-responses-urls.md Shows how to create an HTTP GET request with custom headers. Headers can be provided as a dictionary, and they are internally represented as a list of byte tuples. ```python >>> headers = {"User-Agent": "custom"} >>> request = httpcore.Request("GET", "https://www.example.com/", headers=headers) >>> request.headers [(b'Host', b'www.example.com'), (b"User-Agent", b"custom")] ``` -------------------------------- ### Construct Server-Wide OPTIONS * Requests Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/extensions.md Utilize the 'target' extension to construct server-wide OPTIONS * requests. This is a specific use case for the 'target' extension. ```python # This will send the following request... # # CONNECT * HTTP/1.1 extensions = {"target": b"*"} response = httpx2.request("CONNECT", "https://www.example.com", extensions=extensions) ``` -------------------------------- ### Implementing a Custom Recording Network Backend Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/network-backends.md Create a custom NetworkBackend to record network traffic to a file. This involves implementing RecordingNetworkStream and RecordingNetworkBackend classes. ```python import httpcore class RecordingNetworkStream(httpcore.NetworkStream): def __init__(self, record_file, stream): self.record_file = record_file self.stream = stream def read(self, max_bytes, timeout=None): data = self.stream.read(max_bytes, timeout=timeout) self.record_file.write(data) return data def write(self, buffer, timeout=None): self.stream.write(buffer, timeout=timeout) def close(self) -> None: self.stream.close() def start_tls( self, ssl_context, server_hostname=None, timeout=None, ): self.stream = self.stream.start_tls( ssl_context, server_hostname=server_hostname, timeout=timeout ) return self def get_extra_info(self, info): return self.stream.get_extra_info(info) class RecordingNetworkBackend(httpcore.NetworkBackend): """ A custom network backend that records network responses. """ def __init__(self, record_file): self.record_file = record_file self.backend = httpcore.SyncBackend() def connect_tcp( self, host, port, timeout=None, local_address=None, socket_options=None, ): # Note that we're only using a single record file here, # so even if multiple connections are opened the network # traffic will all write to the same file. # An alternative implementation might automatically use # a new file for each opened connection. stream = self.backend.connect_tcp( host, port, timeout=timeout, local_address=local_address, socket_options=socket_options ) return RecordingNetworkStream(self.record_file, stream) # Once you make the request, the raw HTTP/1.1 response will be available # in the 'network-recording' file. # # Try switching to `http2=True` to see the difference when recording HTTP/2 binary network traffic, # or add `headers={'Accept-Encoding': 'gzip'}` to see HTTP content compression. with open("network-recording", "wb") as record_file: network_backend = RecordingNetworkBackend(record_file) with httpcore.ConnectionPool(network_backend=network_backend) as http: response = http.request("GET", "https://www.example.com/") print(response) ``` -------------------------------- ### HTTP/1.1 Performance Test Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/http2.md Compares the performance of issuing concurrent requests using a standard HTTP/1.1 connection pool. This example demonstrates the number of connections required for parallel requests. ```python import httpcore import concurrent.futures import time def download(http, year): http.request("GET", f"https://en.wikipedia.org/wiki/{year}") def main(): with httpcore.ConnectionPool() as http: started = time.time() with concurrent.futures.ThreadPoolExecutor(max_workers=10) as threads: for year in range(2000, 2020): threads.submit(download, http, year) complete = time.time() for connection in http.connections: print(connection) print("Complete in %.3f seconds" % (complete - started)) main() ``` -------------------------------- ### Making a Request with Timeout Extension Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/extensions.md Demonstrates how to include the 'timeout' extension in a request to specify connection and pool timeouts. ```python r = httpcore.request( "GET", "https://www.example.com", extensions={\"timeout\": {\"connect\": 5.0}} ) ``` -------------------------------- ### Async HTTP Request with TrioBackend Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/network-backends.md Shows how to make an asynchronous HTTP request using the TrioBackend, recommended for applications built with the trio framework for efficiency and cleaner tracebacks. ```python import httpcore import trio async def main(): network_backend = httpcore.TrioBackend() async with httpcore.AsyncConnectionPool(network_backend=network_backend) as http: response = await http.request('GET', 'https://www.example.com') print(response) trio.run(main) ``` -------------------------------- ### Sync and Async Authentication Flows with Locks Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/authentication.md Implement custom authentication that requires custom I/O or concurrency primitives by overriding `sync_auth_flow` and `async_auth_flow`. This example demonstrates using threading and asyncio locks. ```python import asyncio import threading import httpx2 class MyCustomAuth(httpx2.Auth): def __init__(self): self._sync_lock = threading.RLock() self._async_lock = asyncio.Lock() def sync_get_token(self): with self._sync_lock: ... def sync_auth_flow(self, request): token = self.sync_get_token() request.headers["Authorization"] = f"Token {token}" yield request async def async_get_token(self): async with self._async_lock: ... async def async_auth_flow(self, request): token = await self.async_get_token() request.headers["Authorization"] = f"Token {token}" ``` -------------------------------- ### Complex Proxy Routing Configuration Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/transports.md Combines multiple routing rules to establish a sophisticated proxy setup, including default proxies, exclusions, and specific proxy assignments for different domains and subdomains. ```python mounts = { # Route all traffic through a proxy by default... "all://": httpx2.HTTPTransport(proxy="http://localhost:8030"), # But don't use proxies for HTTPS requests to "domain.io"... "https://domain.io": None, # And use another proxy for requests to "example.com" and its subdomains... "all://*example.com": httpx2.HTTPTransport(proxy="http://localhost:8031"), # And yet another proxy if HTTP is used, # and the "internal" subdomain on port 5550 is requested... "http://internal.example.com:5550": httpx2.HTTPTransport(proxy="http://localhost:8032"), } ``` -------------------------------- ### Define No-Proxy Rules for Specific Domains Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/transports.md Configures a default proxy for all requests but excludes requests to 'example.com' from being proxied by setting its mount to None. ```python mounts = { # Route requests through a proxy by default... "all://": httpx2.HTTPTransport(proxy="http://localhost:8031"), # Except those for "example.com". "all://example.com": None, } ``` -------------------------------- ### Default Connection Pool Initialization Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/connection-pools.md Illustrates the default initialization of a ConnectionPool, which is suitable for most use cases as it handles automatic closing upon garbage collection or interpreter exit. ```python # This is perfectly fine for most purposes. # The connection pool will automatically be closed when it is garbage collected, # or when the Python interpreter exits. http = httpcore.ConnectionPool() ``` -------------------------------- ### Trace Extension for Monitoring Transport Events Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/extensions.md Illustrates using the 'trace' extension to install a callback handler for monitoring internal httpcore transport events. The handler receives event names and associated information. ```python import httpx2 def log(event_name, info): print(event_name, info) client = httpx2.Client() response = client.get("https://www.example.com/", extensions={"trace": log}) ``` -------------------------------- ### Concurrent Requests with AnyIO Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Perform multiple HTTP GET requests concurrently using httpcore's AsyncConnectionPool and AnyIO's create_task_group. AnyIO provides a unified API for asyncio and Trio backends. ```python import httpcore import anyio import time async def download(http, year): await http.request("GET", f"https://en.wikipedia.org/wiki/{year}") async def main(): async with httpcore.AsyncConnectionPool() as http: started = time.time() async with anyio.create_task_group() as task_group: for year in range(2000, 2020): task_group.start_soon(download, http, year) complete = time.time() for connection in http.connections: print(connection) print("Complete in %.3f seconds" % (complete - started)) anyio.run(main) ``` -------------------------------- ### Instantiate Connection Pool with HTTP/2 Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/http2.md Create an httpcore ConnectionPool instance with HTTP/2 support enabled. ```python import httpcore pool = httpcore.ConnectionPool(http2=True) ``` -------------------------------- ### Concurrent Requests with AsyncIO Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Send multiple HTTP GET requests concurrently using httpcore's AsyncConnectionPool and asyncio.gather. This is useful for I/O-bound tasks where parallel execution can significantly reduce total request time. ```python import asyncio import httpcore import time async def download(http, year): await http.request("GET", f"https://en.wikipedia.org/wiki/{year}") async def main(): async with httpcore.AsyncConnectionPool() as http: started = time.time() # Here we use `asyncio.gather()` in order to run several tasks concurrently... tasks = [download(http, year) for year in range(2000, 2020)] await asyncio.gather(*tasks) complete = time.time() for connection in http.connections: print(connection) print("Complete in %.3f seconds" % (complete - started)) asyncio.run(main()) ``` -------------------------------- ### Concurrent Requests with Trio Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/async.md Execute multiple HTTP GET requests concurrently using httpcore's AsyncConnectionPool and Trio's open_nursery. Trio's structured concurrency model helps manage concurrent tasks safely. ```python import httpcore import trio import time async def download(http, year): await http.request("GET", f"https://en.wikipedia.org/wiki/{year}") async def main(): async with httpcore.AsyncConnectionPool() as http: started = time.time() async with trio.open_nursery() as nursery: for year in range(2000, 2020): nursery.start_soon(download, http, year) complete = time.time() for connection in http.connections: print(connection) print("Complete in %.3f seconds" % (complete - started)) trio.run(main) ``` -------------------------------- ### Set Connection and Pool Timeouts Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/extensions.md Configure client-level timeouts for connection establishment and waiting on the connection pool. Use the 'timeout' extension for fine-grained control. ```python client = httpx2.Client() response = client.get( "https://www.example.com", extensions={"timeout": {"connect": 5.0, "pool": 10.0}} ) ``` -------------------------------- ### HTTP Transport with Unix Domain Socket (UDS) Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/transports.md Use the `uds` option with HTTPTransport to connect via a Unix Domain Socket, useful for services like the Docker API. The example shows connecting to the Docker API and fetching info. ```pycon >>> import httpx2 >>> # Connect to the Docker API via a Unix Socket. >>> transport = httpx2.HTTPTransport(uds="/var/run/docker.sock") >>> client = httpx2.Client(transport=transport) >>> response = client.get("http://docker/info") >>> response.json() {"ID": "...", "Containers": 4, "Images": 74, ...} ``` -------------------------------- ### Send Request Body on DELETE Method Source: https://github.com/pydantic/httpx2/blob/main/docs/compatibility.md The HTTP GET, DELETE, HEAD, and OPTIONS methods do not typically support request bodies. If you need to send data with these methods in httpx2, use the generic `.request` function instead of specific methods like `.delete`. ```python httpx2.request( method="DELETE", url="https://www.example.com/", content=b'A request body on a DELETE request.' ) ``` -------------------------------- ### HTTP Request with Default Connection Pool Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/network-backends.md Demonstrates making a standard HTTP request using the default connection pool, which automatically selects the network backend. ```python import httpcore with httpcore.ConnectionPool() as http: response = http.request('GET', 'https://www.example.com') print(response) ``` -------------------------------- ### HTTP Request with Explicit SyncBackend Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/network-backends.md Shows how to explicitly select and use the SyncBackend for making an HTTP request with a connection pool. ```python import httpcore network_backend = httpcore.SyncBackend() with httpcore.ConnectionPool(network_backend=network_backend) as http: response = http.request('GET', 'https://www.example.com') print(response) ``` -------------------------------- ### Monitor Download Progress with tqdm Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md Stream a file download and display progress using the tqdm library. This method is recommended for large files as it handles compression correctly. ```python import tempfile import httpx2 from tqdm import tqdm with tempfile.NamedTemporaryFile() as download_file: url = "https://speed.hetzner.de/100MB.bin" with httpx2.stream("GET", url) as response: total = int(response.headers["Content-Length"]) with tqdm(total=total, unit_scale=True, unit_divisor=1024, unit="B") as progress: num_bytes_downloaded = response.num_bytes_downloaded for chunk in response.iter_bytes(): download_file.write(chunk) progress.update(response.num_bytes_downloaded - num_bytes_downloaded) num_bytes_downloaded = response.num_bytes_downloaded ``` -------------------------------- ### Streaming Response Content with AsyncClient Source: https://github.com/pydantic/httpx2/blob/main/docs/async.md Demonstrates how to stream response content using 'client.stream()' and iterate over chunks asynchronously. ```pycon >>> client = httpx2.AsyncClient() >>> async with client.stream('GET', 'https://www.example.com/') as response: ... async for chunk in response.aiter_bytes(): ... ... ``` -------------------------------- ### Observe Connection Pooling Benefits Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/connection-pools.md This script illustrates the performance benefits of connection pooling by sending multiple requests to the same host and measuring the time taken for each. It highlights how subsequent requests are faster due to connection reuse. ```python import httpcore import time http = httpcore.ConnectionPool() for counter in range(5): started = time.time() response = http.request("GET", "https://www.example.com/") complete = time.time() print(response, "in %.3f seconds" % (complete - started)) ``` -------------------------------- ### Configure Client with Certifi CA Bundle Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/ssl.md Configures an httpx2 client to use the certifi CA bundle for SSL verification instead of the system trust store. ```python import certifi import httpx2 import ssl # Use the certifi CA bundle. c_tx = ssl.create_default_context(cafile=certifi.where()) client = httpx2.Client(verify=c_tx) ``` -------------------------------- ### Use AsyncClient as a context manager with HTTP/2 Source: https://github.com/pydantic/httpx2/blob/main/docs/http2.md Instantiate an httpx2.AsyncClient with HTTP/2 enabled as a context manager to ensure connections are properly closed. ```python async with httpx2.AsyncClient(http2=True) as client: ... ``` -------------------------------- ### Send a Request Instance with a Client Source: https://github.com/pydantic/httpx2/blob/main/docs/advanced/clients.md Use a Client instance to send a pre-built Request object over the network. This pattern is useful for reusing client configurations. ```python with httpx2.Client() as client: response = client.send(request) ... ``` -------------------------------- ### HTTPX Client Instance vs. Requests Session Source: https://github.com/pydantic/httpx2/blob/main/docs/compatibility.md The `httpx2.Client` class is the equivalent of `requests.Session` for managing client state and configuration. ```python session = requests.Session(**kwargs) ``` ```python client = httpx2.Client(**kwargs) ``` -------------------------------- ### Connection Pool Lifecycle Management (Context Manager) Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/connection-pools.md Shows how to manage the lifecycle of a ConnectionPool instance using a context manager, ensuring resources are properly released. ```python with httpcore.ConnectionPool() as http: ... ``` -------------------------------- ### Constructing a Request Object Source: https://github.com/pydantic/httpx2/blob/main/docs/api.md Demonstrates how to manually construct a httpx2.Request object for detailed control over outgoing HTTP requests. This is useful when you need to specify exact headers or other request details before sending. ```python >>> request = httpx2.Request("GET", "https://example.org", headers={'host': 'example.org'}) >>> response = client.send(request) ``` -------------------------------- ### Mocking a Simple HTTP/1.1 Response Source: https://github.com/pydantic/httpx2/blob/main/src/httpcore2/docs/network-backends.md Use MockBackend to simulate a basic HTTP/1.1 response. This is useful for testing client logic without making actual network requests. ```python import httpcore network_backend = httpcore.MockBackend([ b"HTTP/1.1 200 OK\r\n", b"Content-Type: plain/text\r\n", b"Content-Length: 13\r\n", b"\r\n", b"Hello, world!", ]) with httpcore.ConnectionPool(network_backend=network_backend) as http: response = http.request("GET", "https://example.com/") print(response.extensions['http_version']) print(response.status) print(response.content) ```