Interceptors & HTTP

Interceptors¶

Interceptors let you hook into the operation lifecycle. An operation interceptor receives the operation request and a next callable that invokes the rest of the chain, returning the operation response. You can inspect or modify the request before calling next, and inspect or modify the response after.

import asyncio
from typing import Any, Awaitable, Callable

from capo_s3 import AsyncOperationRequest, AsyncOperationResponse, AsyncS3Client


async def debug_interceptor(
    request: AsyncOperationRequest[Any],
    next: Callable[[AsyncOperationRequest[Any]], Awaitable[AsyncOperationResponse]],
):
    print(request)
    response = await next(request)
    print(response)
    return response


async def main():
    async with AsyncS3Client(operation_interceptors=[debug_interceptor]) as client:
        async for item in client.iter_list_buckets():
            print(item)


asyncio.run(main())

Note An operation_interceptor sits between the operation request and the operation response — not at the HTTP layer. For HTTP-level interceptors/middlewares, see the Configuring the HTTP client section below.

Configuring the HTTP client¶

The SDK is built on zapros. You can pass your own HTTP handler via the http_handler argument, including any zapros middleware wrapping a network handler. This is the right place for HTTP-level concerns such as caching, mocks, or custom transports. See the zapros handlers documentation for the available handlers and middlewares.

import asyncio

from zapros import AsyncStdNetworkHandler, CacheMiddleware

from capo_s3 import AsyncS3Client


async def main():
    async with AsyncS3Client(
        http_handler=CacheMiddleware(AsyncStdNetworkHandler())
    ) as client:
        async for item in client.iter_list_buckets():
            print(item)


asyncio.run(main())

Retrying¶

The SDK retries failed operations automatically. Retry behaviour follows the Smithy specification: errors are retried based on their is_retryable and is_throttling_error attributes. Throttling errors use a longer base delay. Network-level failures (connection errors and timeouts) are also retried. Non-retryable errors, such as client errors without the @retryable trait, are raised immediately without further attempts.

The number of attempts defaults to 3 and can be changed at the client level via retry_max_attempts, or per call via config_overrides.

from capo_s3 import AsyncS3Client


async def main():
    async with AsyncS3Client() as s3:
        # Default: 3 attempts for every operation
        response = await s3.abort_multipart_upload()

        # Override per operation
        response = await s3.abort_multipart_upload(config_overrides={"retry_max_attempts": 5})

        # Disable retries for this call
        response = await s3.abort_multipart_upload(config_overrides={"retry_max_attempts": 1})