Advanced Usage — Aquilia Documentation
Comprehensive guide and documentation for Advanced Usage in the Aquilia framework. View API reference, examples, and implementation patterns.
HTTP Client / Advanced Advanced Usage Advanced HTTP client features: streaming, retry strategies, interceptors, and middleware. Streaming Responses Process large response payloads without loading the entire body into memory: client = AsyncHTTPClient() # Stream response body bytes response = await client.get("https://example.com/large-file.csv") async for chunk in response.iter_bytes(chunk_size=8192): # Process chunk (bytes) process_data(chunk) # Stream to file response = await client.get("https://example.com/video.mp4") with open("video.mp4", "wb") as f: async for chunk in response.iter_bytes(): f.write(chunk) # Stream lines (text) response = await client.get("https://example.com/logs.txt") async for line in response.iter_lines(): # Each line is decoded as UTF-8 print(f"Log: ") # Stream JSON lines (JSONL/NDJSON) response = await client.get("https://api.example.com/stream") async for line in response.iter_lines(): if line.strip(): record = json.loads(line) process_record(record) Streaming Requests Stream large file uploads or generators to minimize memory footprint: # Stream file upload without loading into memory async def file_stream(): with open("large-file.bin", "rb") as f: while chunk := f.read(1024 * 64): yield chunk response = await client.post( "/upload", data=file_stream(), headers= , ) # Stream generated data async def data_generator(): for i in range(100): yield f"Line \n".encode() response = await client.post("/stream-upload", data=data_generator()) Retry Strategies Configure automatic retries with exponential backoff for transient failures using RetryConfig : from aquilia.http import AsyncHTTPClient, RetryConfig client = AsyncHTTPClient(HTTPClientConfig( retry=RetryConfig( max_attempts=3, # Retry up to 3 times backoff_base=1.0, # Exponential backoff base delay (seconds) backoff_max=30.0, # Cap backoff delay retry_on_status= , ), )) Interceptors Hook actions that execute before a request is sent, or after a response is received: from aquilia.http.interceptors import HTTPInterceptor # Logging interceptor class LoggingInterceptor(HTTPInterceptor): async def intercept(self, request, handler): print(f"→ ") response = await handler(request) print(f"← ") return response client = AsyncHTTPClient(interceptors=[ LoggingInterceptor(), ]) Concurrent Requests import asyncio client = AsyncHTTPClient() urls = [ "https://api.example.com/users/1", "https://api.example.com/users/2", ] responses = await asyncio.gather(*[ client.get(url) for url in urls ]) # Limit concurrency with semaphore semaphore = asyncio.Semaphore(5) async def fetch(url: str): async with semaphore: return await client.get(url) Proxy Support # HTTP proxy client = AsyncHTTPClient(HTTPClientConfig( proxy=ProxyConfig(http_proxy="http://proxy.example.com:8080"), )) # Environment variable fallback import os os.environ["HTTP_PROXY"] = "http://proxy.example.com:8080" client = AsyncHTTPClient() Transport Layer Error Handling )
Go to Homepage