1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320
|
#!/usr/bin/env python3
"""
Example of combining multiple middleware with aiohttp client.
This example shows how to chain multiple middleware together to create
a powerful request pipeline. Middleware are applied in order, demonstrating
how logging, authentication, and retry logic can work together.
The order of middleware matters:
1. Logging (outermost) - logs all attempts including retries
2. Authentication - adds auth headers before retry logic
3. Retry (innermost) - retries requests on failure
"""
import asyncio
import base64
import binascii
import logging
import time
from http import HTTPStatus
from typing import TYPE_CHECKING, Set, Union
from aiohttp import (
ClientHandlerType,
ClientRequest,
ClientResponse,
ClientSession,
hdrs,
web,
)
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
_LOGGER = logging.getLogger(__name__)
class LoggingMiddleware:
"""Middleware that logs request timing and response status."""
async def __call__(
self,
request: ClientRequest,
handler: ClientHandlerType,
) -> ClientResponse:
start_time = time.monotonic()
# Log request
_LOGGER.info("[REQUEST] %s %s", request.method, request.url)
# Execute request
response = await handler(request)
# Log response
duration = time.monotonic() - start_time
_LOGGER.info(
"[RESPONSE] %s in %.2fs - Status: %s",
request.url.path,
duration,
response.status,
)
return response
class BasicAuthMiddleware:
"""Middleware that adds Basic Authentication to all requests."""
def __init__(self, username: str, password: str) -> None:
self.username = username
self.password = password
self._auth_header = self._encode_credentials()
def _encode_credentials(self) -> str:
"""Encode username and password to base64."""
credentials = f"{self.username}:{self.password}"
encoded = base64.b64encode(credentials.encode()).decode()
return f"Basic {encoded}"
async def __call__(
self,
request: ClientRequest,
handler: ClientHandlerType,
) -> ClientResponse:
"""Add Basic Auth header to the request."""
# Only add auth if not already present
if hdrs.AUTHORIZATION not in request.headers:
request.headers[hdrs.AUTHORIZATION] = self._auth_header
_LOGGER.debug("Added Basic Auth header")
# Proceed with the request
return await handler(request)
DEFAULT_RETRY_STATUSES: Set[HTTPStatus] = {
HTTPStatus.TOO_MANY_REQUESTS,
HTTPStatus.INTERNAL_SERVER_ERROR,
HTTPStatus.BAD_GATEWAY,
HTTPStatus.SERVICE_UNAVAILABLE,
HTTPStatus.GATEWAY_TIMEOUT,
}
class RetryMiddleware:
"""Middleware that retries failed requests with exponential backoff."""
def __init__(
self,
max_retries: int = 3,
retry_statuses: Union[Set[HTTPStatus], None] = None,
initial_delay: float = 1.0,
backoff_factor: float = 2.0,
) -> None:
self.max_retries = max_retries
self.retry_statuses = retry_statuses or DEFAULT_RETRY_STATUSES
self.initial_delay = initial_delay
self.backoff_factor = backoff_factor
async def __call__(
self,
request: ClientRequest,
handler: ClientHandlerType,
) -> ClientResponse:
"""Execute request with retry logic."""
last_response: Union[ClientResponse, None] = None
delay = self.initial_delay
for attempt in range(self.max_retries + 1):
if attempt > 0:
_LOGGER.info(
"Retrying request (attempt %s/%s)",
attempt + 1,
self.max_retries + 1,
)
# Execute the request
response = await handler(request)
last_response = response
# Check if we should retry
if response.status not in self.retry_statuses:
return response
# Don't retry if we've exhausted attempts
if attempt >= self.max_retries:
_LOGGER.warning("Max retries exceeded")
return response
# Wait before retrying
_LOGGER.debug("Waiting %ss before retry...", delay)
await asyncio.sleep(delay)
delay *= self.backoff_factor
if TYPE_CHECKING:
assert last_response is not None # Always set since we loop at least once
return last_response
class TestServer:
"""Test server with stateful endpoints for middleware testing."""
def __init__(self) -> None:
self.flaky_counter = 0
self.protected_counter = 0
async def handle_protected(self, request: web.Request) -> web.Response:
"""Protected endpoint that requires authentication and is flaky on first attempt."""
auth_header = request.headers.get(hdrs.AUTHORIZATION, "")
if not auth_header.startswith("Basic "):
return web.Response(
status=401,
text="Unauthorized",
headers={hdrs.WWW_AUTHENTICATE: 'Basic realm="test"'},
)
# Decode the credentials
encoded_creds = auth_header[6:] # Remove "Basic "
try:
decoded = base64.b64decode(encoded_creds).decode()
username, password = decoded.split(":", 1)
except (ValueError, binascii.Error):
return web.Response(
status=401,
text="Invalid credentials format",
headers={hdrs.WWW_AUTHENTICATE: 'Basic realm="test"'},
)
# Validate credentials
if username != "user" or password != "pass":
return web.Response(status=401, text="Invalid credentials")
# Fail with 500 on first attempt to test retry + auth combination
self.protected_counter += 1
if self.protected_counter == 1:
return web.Response(
status=500, text="Internal server error (first attempt)"
)
return web.json_response(
{
"message": "Access granted",
"user": username,
"resource": "protected data",
}
)
async def handle_flaky(self, request: web.Request) -> web.Response:
"""Endpoint that fails a few times before succeeding."""
self.flaky_counter += 1
# Fail the first 2 requests, succeed on the 3rd
if self.flaky_counter <= 2:
return web.Response(
status=503,
text=f"Service temporarily unavailable (attempt {self.flaky_counter})",
)
# Reset counter and return success
self.flaky_counter = 0
return web.json_response(
{
"message": "Success after retries!",
"data": "Important information retrieved",
}
)
async def handle_always_fail(self, request: web.Request) -> web.Response:
"""Endpoint that always returns an error."""
return web.Response(status=500, text="Internal server error")
async def handle_status(self, request: web.Request) -> web.Response:
"""Return the status code specified in the path."""
status = int(request.match_info["status"])
return web.Response(status=status, text=f"Status: {status}")
async def run_test_server() -> web.AppRunner:
"""Run a test server with various endpoints."""
app = web.Application()
server = TestServer()
app.router.add_get("/protected", server.handle_protected)
app.router.add_get("/flaky", server.handle_flaky)
app.router.add_get("/always-fail", server.handle_always_fail)
app.router.add_get("/status/{status}", server.handle_status)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "localhost", 8080)
await site.start()
return runner
async def run_tests() -> None:
"""Run all the middleware tests."""
# Create middleware instances
logging_middleware = LoggingMiddleware()
auth_middleware = BasicAuthMiddleware("user", "pass")
retry_middleware = RetryMiddleware(max_retries=2, initial_delay=0.5)
# Combine middleware - order matters!
# Applied in order: logging -> auth -> retry -> request
async with ClientSession(
middlewares=(logging_middleware, auth_middleware, retry_middleware)
) as session:
print(
"=== Test 1: Protected endpoint with auth (fails once, then succeeds) ==="
)
print("This tests retry + auth working together...")
async with session.get("http://localhost:8080/protected") as resp:
if resp.status == 200:
data = await resp.json()
print(f"Success after retry! Response: {data}")
else:
print(f"Failed with status: {resp.status}")
print("\n=== Test 2: Flaky endpoint (fails twice, then succeeds) ===")
print("Watch the logs to see retries in action...")
async with session.get("http://localhost:8080/flaky") as resp:
if resp.status == 200:
data = await resp.json()
print(f"Success after retries! Response: {data}")
else:
text = await resp.text()
print(f"Failed with status {resp.status}: {text}")
print("\n=== Test 3: Always failing endpoint ===")
async with session.get("http://localhost:8080/always-fail") as resp:
print(f"Final status after retries: {resp.status}")
print("\n=== Test 4: Non-retryable status (404) ===")
async with session.get("http://localhost:8080/status/404") as resp:
print(f"Status: {resp.status} (no retries for 404)")
# Test without middleware for comparison
print("\n=== Test 5: Request without middleware ===")
print("Making a request to protected endpoint without middleware...")
async with session.get(
"http://localhost:8080/protected", middlewares=()
) as resp:
print(f"Status without middleware: {resp.status}")
if resp.status == 401:
print("Failed as expected - no auth header added")
async def main() -> None:
# Start test server
server = await run_test_server()
try:
await run_tests()
finally:
await server.cleanup()
if __name__ == "__main__":
asyncio.run(main())
|