HTTPClient — Aquilia Documentation
Comprehensive guide and documentation for HTTPClient in the Aquilia framework. View API reference, examples, and implementation patterns.
HTTP Client / Basics HTTPClient The AsyncHTTPClient class provides a high-level async API for making HTTP requests with built-in retry logic, interceptors, and connection pooling. Basic Usage from aquilia.http import AsyncHTTPClient async def main(): # Create client client = AsyncHTTPClient() try: # Make GET request response = await client.get("https://api.github.com/users/octocat") # Check status if response.is_success: data = await response.json() print(f"User: ") # Response provides helpers print(f"Status: ") print(f"Headers: ") print(f"Elapsed: s") finally: # Always close to release connections await client.close() # Or use context manager async def with_context(): async with AsyncHTTPClient() as client: response = await client.get("https://httpbin.org/get") data = await response.json() return data Configuration AsyncHTTPClient supports comprehensive configuration via HTTPClientConfig . All nested configurations are fully configurable: from aquilia.http import ( AsyncHTTPClient, HTTPClientConfig, TimeoutConfig, PoolConfig, RetryConfig, ProxyConfig, TLSConfig ) client = AsyncHTTPClient(HTTPClientConfig( # Base URL (prepended to all relative URLs) base_url="https://api.example.com", # Timeout configuration with granular control timeout=TimeoutConfig( total=30.0, # Overall request timeout (None = no limit) connect=10.0, # Connection establishment timeout read=20.0, # Read timeout for response data write=10.0, # Write timeout for request data pool=5.0, # Pool acquisition timeout ), # Connection pool configuration pool=PoolConfig( max_connections=100, # Global connection limit max_connections_per_host=10, # Per-host connection limit keepalive_expiry=60.0, # Keep-alive duration (seconds) ), # Retry configuration with exponential backoff retry=RetryConfig( max_attempts=3, # Number of retry attempts backoff_base=1.0, # Base delay for backoff backoff_multiplier=2.0, # Exponential backoff multiplier backoff_max=60.0, # Maximum backoff delay backoff_jitter=0.1, # Add randomness to backoff retry_on_status= , # Retry on these status codes retry_on_methods= , # Retry these methods (idempotent) ), # Proxy configuration (supports environment variables) proxy=ProxyConfig( http_proxy="http://proxy.corp:8080", https_proxy="https://proxy.corp:8080", no_proxy="localhost,127.0.0.1,*.local", ), # TLS/SSL configuration tls=TLSConfig( verify=True, # Verify SSL certificates cert_file="/path/to/client.crt", # Client certificate key_file="/path/to/client.key", # Client private key ca_bundle="/path/to/ca.pem", # Custom CA bundle ciphers="ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM", # Cipher suite ), # Default headers for all requests default_headers= , # Redirect handling follow_redirects=True, # Follow 3xx redirects max_redirects=10, # Maximum redirect hops # Error handling raise_for_status=False, # Raise HTTPStatusFault for 4xx/5xx # User agent override user_agent="Aquilia-HTTP/1.0 MyApp/2.1", )) Configuration Presets Use factory methods on TimeoutConfig and RetryConfig for common scenarios: # Fast config for internal APIs config = HTTPClientConfig( timeout=TimeoutConfig.fast(), # total=5.0, connect=2.0 retry=RetryConfig.no_retry(), # max_attempts=0 ) # Slow config for external APIs config = HTTPClientConfig( timeout=TimeoutConfig.slow(), # total=60.0, connect=15.0 retry=RetryConfig.aggressive(), # max_attempts=5, backoff_max=30.0 ) # No timeout for long downloads config = HTTPClientConfig( timeout=TimeoutConfig.no_timeout(), # All timeouts = None ) # Serialize/deserialize config config_dict = config.to_dict() restored = HTTPClientConfig.from_dict(config_dict) Request Methods AsyncHTTPClient provides convenience methods for all HTTP verbs: client = AsyncHTTPClient() # GET request response = await client.get("/users") # POST with JSON body response = await client.post( "/users", json= ) # PUT with custom headers response = await client.put( "/users/123", json= , headers= ) # PATCH response = await client.patch("/users/123", json= ) # DELETE response = await client.delete("/users/123") # HEAD (no body) response = await client.head("/health") # OPTIONS response = await client.options("/api") # Generic request method response = await client.request( "GET", "/custom", headers= , params= , ) Request Parameters All request methods accept these parameters to customize the outgoing request: ))} Query Parameters # Query params as dict response = await client.get( "/search", params= ) # Sends: GET /search?q=python&page=1&limit=10 # URL encoding handled automatically response = await client.get( "/search", params= ) # Sends: GET /search?q=async+programming&tags=python%2Casyncio # Array-like params params = response = await client.get("/users", params=params) # Sends: GET /users?filter=active&filter=verified Headers # Per-request headers response = await client.get( "/api/data", headers= ) # Headers are case-insensitive response = await client.get( "/api/data", headers= # Same as "Accept" ) # Merge with default headers client = AsyncHTTPClient(HTTPClientConfig( default_headers= )) # Request headers merge with defaults response = await client.get("/", headers= ) # Sends: User-Agent: MyApp/1.0, X-Custom: value JSON Requests # JSON is auto-serialized response = await client.post( "/api/users", json= } ) # Automatically sets: # - Content-Type: application/json # - Serializes dict to JSON string # Complex data structures import datetime data = , } response = await client.post("/api/records", json=data) File Uploads and Multipart Forms Supports robust multipart form building with progress callbacks and automatic content type detection via MultipartFormData : from aquilia.http import MultipartFormData from pathlib import Path # Using MultipartFormData builder form = ( MultipartFormData() .field("title", "My Document") .field("description", "A sample file upload") .file("document", "report.pdf", open("report.pdf", "rb")) .file_from_path("image", Path("chart.png")) .file_from_bytes("config", "config.json", b' ', "application/json") ) response = await client.post("/upload", files=form) # Progress tracking with callbacks async def progress_callback(progress): print(f"Uploaded: / " f"( %) at B/s") # Streaming upload for large files from aquilia.http.streaming import StreamingBody stream = StreamingBody( Path("large_file.zip"), chunk_size=65536, on_progress=progress_callback ) form = MultipartFormData().file("upload", "large_file.zip", stream) response = await client.post("/upload", files=form) Authentication Supports 6 built-in authentication schemes implemented as request interceptors: Basic Authentication (RFC 7617) from aquilia.http import BasicAuth auth = BasicAuth("username", "password") client = AsyncHTTPClient(interceptors=[auth]) # Or per-request response = await client.get("/protected", auth=("user", "pass")) Bearer Token Authentication from aquilia.http import BearerAuth # Static token auth = BearerAuth("your-jwt-token") # Dynamic token with callback async def get_token(): return "dynamic-resolved-token" auth = BearerAuth(get_token) # Resolves for each request client = AsyncHTTPClient(interceptors=[auth]) API Key Authentication from aquilia.http import APIKeyAuth # Header-based API key auth = APIKeyAuth("X-API-Key", "your-api-key", location="header") # Query parameter API key auth = APIKeyAuth("api_key", "your-api-key", location="query") client = AsyncHTTPClient(interceptors=[auth]) Digest Authentication (RFC 7616) from aquilia.http import DigestAuth # Automatic challenge-response workflow auth = DigestAuth("username", "password") client = AsyncHTTPClient(interceptors=[auth]) OAuth2 Authentication from aquilia.http import OAuth2Auth auth = OAuth2Auth( client_id="your-client-id", client_secret="your-client-secret", token_url="https://auth.provider.com/oauth/token", initial_token="current-access-token", ) client = AsyncHTTPClient(interceptors=[auth]) AWS Signature V4 Authentication from aquilia.http import AWSSignatureV4Auth auth = AWSSignatureV4Auth( access_key="ACCESS_KEY_EXAMPLE", secret_key="SECRET_KEY_EXAMPLE", region="us-east-1", service="s3" ) client = AsyncHTTPClient(interceptors=[auth]) Streaming Requests and Responses Stream large file uploads or response payloads to minimize memory footprints: Streaming Response Data # Stream response in chunks response = await client.get("https://example.com/large-file.zip") async for chunk in response.iter_bytes(chunk_size=8192): process_chunk(chunk) # Stream response as text lines response = await client.get("https://api.example.com/logs") async for line in response.iter_lines(): parse_log_line(line) # Stream response as text with encoding response = await client.get("https://example.com/data.csv") async for chunk in response.iter_text(chunk_size=4096, encoding="utf-8"): parse_csv_chunk(chunk) Streaming Request Bodies from aquilia.http.streaming import StreamingBody # Stream file upload stream = StreamingBody( Path("large-video.mp4"), chunk_size=65536, ) response = await client.put("/upload/video", data=stream) # Stream from async generator async def generate_data(): for i in range(100): yield f"chunk- \n".encode() stream = StreamingBody(generate_data(), chunk_size=1024) response = await client.post("/stream-data", data=stream) Fault Hierarchy All errors thrown by the HTTP client inherit from HTTPClientFault: HTTPClientFault — Base exception class. ├─ ConnectionFault — TCP/connection handshake errors. ├─ TimeoutFault — Granular Connect/Read/Write timeout errors. │ ├─ ConnectTimeoutFault │ ├─ ReadTimeoutFault │ └─ WriteTimeoutFault ├─ TLSFault — SSL Handshake and cert verification failures. ├─ HTTPStatusFault — Non-2xx status codes (when raise_for_status is True). │ ├─ ClientErrorFault (4xx) │ └─ ServerErrorFault (5xx) ├─ RedirectFault — Redirection loop or limit exceeded. └─ RetryExhaustedFault — Exceeded max retry counts. Overview Sessions )
Go to Homepage