4. Dependency Injection for boto3 Clients

Overview

In this capsule you'll implement dependency injection for boto3 clients. Instead of your business logic building its own S3 or Lambda clients (deciding endpoints, credentials, region), it will receive them already configured. A factory pattern creates the clients according to the environment, and your business code simply uses them. It's the difference between a function that knows too much about infrastructure and a function that only knows how to process documents.

Context: In capsule 02 you abstracted the environment. In 03 you built typed config management. Now you'll connect the two: the factory reads the configuration, builds correctly configured boto3 clients, and injects them where they're needed. Your DocumentProcessor won't import boto3, won't read environment variables, won't decide whether it uses LocalStack or AWS. It receives a client and works. If tomorrow you switch from S3 to MinIO, you only change the factory — the processor never even notices.

Why Dependency Injection for Cloud Clients

The anti-pattern: business logic that builds clients

# ❌ Anti-pattern — the business logic knows too much
import boto3
import os


class DocumentProcessor:
    def __init__(self):
        endpoint = os.environ.get("AWS_ENDPOINT_URL")
        self.s3 = boto3.client(
            "s3",
            endpoint_url=endpoint,
            aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
            aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
            region_name=os.environ.get("AWS_REGION", "us-east-1"),
        )
        self.bucket = os.environ.get("S3_BUCKET", "default")

    def process(self, key: str) -> dict:
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        # ... processing ...

Problems:

• ❌ DocumentProcessor knows infrastructure details (endpoint_url, credentials)
• ❌ Impossible to test with a mock S3 without manipulating environment variables
• ❌ Every class that needs S3 repeats the same client construction
• ❌ Switching providers (S3 → MinIO → GCS) requires editing business logic

The pattern: inject configured dependencies

# ✅ Correct pattern — the business logic is pure
class DocumentProcessor:
    def __init__(self, s3_client, bucket_name: str):
        self.s3 = s3_client
        self.bucket = bucket_name

    def process(self, key: str) -> dict:
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        # ... processing ...

DocumentProcessor doesn't know what boto3 is. It doesn't know whether the client talks to LocalStack or AWS. It doesn't import os. It receives a client that has .get_object() and a bucket name — and it works.

Concrete benefits

Without DI                           With DI
──────────────────────────────       ──────────────────────────────
Class creates its client              Class receives a client
  → coupled to boto3                    → works with any client
  → reads env vars                      → doesn't depend on env vars
  → hard to test                        → easy to test (mock)
  → changing provider = rewrite        → changing provider = change factory
  → N classes × repeated construction  → 1 factory × N classes

The Factory Pattern for boto3

ClientFactory: centralized client creation

"""clients/factory.py — Factory to create boto3 clients."""

import boto3
from typing import Any
from config.settings import Settings


class ClientFactory:
    """Creates boto3 clients configured for the current environment.

    Centralizes the construction logic: endpoints, credentials,
    region. No other module needs to know how boto3 is configured.
    """

    def __init__(self, settings: Settings):
        self.settings = settings
        self._base_kwargs = self._build_base_kwargs()
        self._clients: dict[str, Any] = {}

    def _build_base_kwargs(self) -> dict:
        """Builds common kwargs for all clients."""
        kwargs = {"region_name": self.settings.aws_region}

        if self.settings.aws_endpoint_url:
            kwargs["endpoint_url"] = self.settings.aws_endpoint_url

        if self.settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = self.settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = self.settings.aws_secret_access_key

        return kwargs

    def get_client(self, service: str) -> Any:
        """Returns a boto3 client for the given service.

        Clients are cached: the same service returns the same instance.
        """
        if service not in self._clients:
            self._clients[service] = boto3.client(service, **self._base_kwargs)
        return self._clients[service]

    @property
    def s3(self) -> Any:
        return self.get_client("s3")

    @property
    def lambda_client(self) -> Any:
        return self.get_client("lambda")

    @property
    def logs(self) -> Any:
        return self.get_client("logs")

    @property
    def secretsmanager(self) -> Any:
        return self.get_client("secretsmanager")

    def health_check(self) -> dict:
        """Verifies connectivity of all cached clients."""
        results = {}
        for service, client in self._clients.items():
            try:
                if service == "s3":
                    client.list_buckets()
                elif service == "lambda":
                    client.list_functions(MaxItems=1)
                elif service == "logs":
                    client.describe_log_groups(limit=1)
                results[service] = "healthy"
            except Exception as e:
                results[service] = f"unhealthy: {e}"
        return results

    def __repr__(self) -> str:
        return (
            f"ClientFactory(env={self.settings.environment}, "
            f"endpoint={self.settings.aws_endpoint_url or 'AWS'}, "
            f"cached_clients={list(self._clients.keys())})"
        )

Using the factory

from config.loader import get_settings
from clients.factory import ClientFactory

settings = get_settings()
factory = ClientFactory(settings)

# The clients are configured for the right environment
s3 = factory.s3
s3.list_buckets()  # Talks to LocalStack or AWS depending on ENVIRONMENT

lambda_client = factory.lambda_client
lambda_client.list_functions()

print(factory)
# ClientFactory(env=local, endpoint=http://localhost:4566, cached_clients=['s3', 'lambda'])

Injecting Clients into Services

DocumentProcessor with DI

"""services/document_processor.py — Document processor with DI."""

import json
from datetime import datetime
from typing import Any


class DocumentProcessor:
    """Processes AI documents.

    It doesn't know boto3, doesn't know the environment, doesn't read environment variables.
    It receives clients and configuration — and works.
    """

    def __init__(self, s3_client: Any, bucket: str):
        self.s3 = s3_client
        self.bucket = bucket

    def get_prompt(self, name: str, version: str) -> str:
        key = f"prompts/{name}/{version}/system.txt"
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        return response["Body"].read().decode("utf-8")

    def store_document(self, collection: str, doc_id: str, data: dict) -> str:
        key = f"documents/{collection}/{doc_id}.json"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(data, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

    def save_response(self, request_id: str, result: dict) -> str:
        now = datetime.utcnow()
        key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
        payload = {"request_id": request_id, "timestamp": now.isoformat(), **result}
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

InferenceService with DI

"""services/inference.py — Inference service with DI."""

import json
from typing import Any, Optional


class InferenceService:
    """Runs inference by invoking a Lambda function.

    It receives a lambda client and the function name.
    It doesn't know whether the Lambda runs on LocalStack or AWS.
    """

    def __init__(self, lambda_client: Any, function_name: str):
        self.lambda_client = lambda_client
        self.function_name = function_name

    def invoke(self, payload: dict) -> dict:
        response = self.lambda_client.invoke(
            FunctionName=self.function_name,
            InvocationType="RequestResponse",
            Payload=json.dumps(payload).encode("utf-8"),
        )

        status_code = response["StatusCode"]
        response_payload = json.loads(
            response["Payload"].read().decode("utf-8")
        )

        if status_code != 200:
            raise RuntimeError(
                f"Lambda returned status {status_code}: {response_payload}"
            )

        return response_payload

    def invoke_async(self, payload: dict) -> str:
        response = self.lambda_client.invoke(
            FunctionName=self.function_name,
            InvocationType="Event",
            Payload=json.dumps(payload).encode("utf-8"),
        )
        return response["StatusCode"]

Wiring: connecting the factory with the services

"""app.py — Wiring: connects config, factory, and services."""

from config.loader import get_settings
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor
from services.inference import InferenceService


def create_app():
    """Builds the application with all dependencies injected."""
    settings = get_settings()
    factory = ClientFactory(settings)

    processor = DocumentProcessor(
        s3_client=factory.s3,
        bucket=settings.s3_bucket,
    )

    inference = InferenceService(
        lambda_client=factory.lambda_client,
        function_name=settings.lambda_function_name,
    )

    return {
        "settings": settings,
        "factory": factory,
        "processor": processor,
        "inference": inference,
    }


# Usage
app = create_app()
print(f"Environment: {app['settings'].environment}")
print(f"Factory: {app['factory']}")

prompt = app["processor"].get_prompt("summarizer", "v1")
print(f"Prompt loaded: {prompt[:60]}...")

Change ENVIRONMENT=local to ENVIRONMENT=staging and the same create_app() builds everything against AWS. Zero changes to DocumentProcessor, InferenceService, or any other service.

Service Container: Advanced Organization

A container that manages all the services

"""services/container.py — Service container with DI."""

from dataclasses import dataclass
from typing import Any
from config.settings import Settings
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor
from services.inference import InferenceService


@dataclass
class ServiceContainer:
    """Holds all the application's services, already injected."""
    settings: Settings
    factory: ClientFactory
    processor: DocumentProcessor
    inference: InferenceService

    @classmethod
    def create(cls, settings: Settings | None = None) -> "ServiceContainer":
        """Factory method that builds the complete container."""
        if settings is None:
            from config.loader import get_settings
            settings = get_settings()

        factory = ClientFactory(settings)

        processor = DocumentProcessor(
            s3_client=factory.s3,
            bucket=settings.s3_bucket,
        )

        inference = InferenceService(
            lambda_client=factory.lambda_client,
            function_name=settings.lambda_function_name,
        )

        return cls(
            settings=settings,
            factory=factory,
            processor=processor,
            inference=inference,
        )

    def health(self) -> dict:
        """Health check of all the services."""
        return {
            "environment": self.settings.environment,
            "clients": self.factory.health_check(),
        }

Use in a Lambda handler

"""handler.py — Lambda handler with DI via ServiceContainer."""

import json
from services.container import ServiceContainer

container: ServiceContainer | None = None


def get_container() -> ServiceContainer:
    """Lazy initialization of the container (reused between invocations)."""
    global container
    if container is None:
        container = ServiceContainer.create()
    return container


def lambda_handler(event, context):
    """Lambda handler with injected dependencies."""
    c = get_container()

    path = event.get("rawPath", event.get("path", ""))
    method = event.get("requestContext", {}).get("http", {}).get("method", "GET")

    if path == "/health" and method == "GET":
        health = c.health()
        return {
            "statusCode": 200,
            "body": json.dumps(health),
        }

    if path == "/process" and method == "POST":
        body = json.loads(event.get("body", "{}"))
        prompt_name = body.get("prompt_name", "summarizer")
        prompt_version = body.get("prompt_version", "v1")
        document = body.get("document", {})

        prompt = c.processor.get_prompt(prompt_name, prompt_version)
        stored_key = c.processor.store_document(
            "inbox", document.get("id", "unknown"), document
        )
        result = {
            "prompt_used": f"{prompt_name}/{prompt_version}",
            "document_stored": stored_key,
            "processed": True,
        }
        response_key = c.processor.save_response(
            f"req-{context.aws_request_id[:8] if context else 'local'}",
            result,
        )
        result["response_key"] = response_key

        return {
            "statusCode": 200,
            "body": json.dumps(result),
        }

    return {
        "statusCode": 404,
        "body": json.dumps({"error": "Not found"}),
    }

Testing with DI: The Real Advantage

Mock clients for unit tests

The most practical advantage of DI: testing without LocalStack or AWS.

"""tests/test_document_processor.py — Tests with mock clients."""

import json
from io import BytesIO
from unittest.mock import MagicMock
from services.document_processor import DocumentProcessor


def make_s3_response(content: str) -> dict:
    """Helper: creates a mocked S3 response."""
    body = MagicMock()
    body.read.return_value = content.encode("utf-8")
    return {"Body": body}


def test_get_prompt():
    mock_s3 = MagicMock()
    mock_s3.get_object.return_value = make_s3_response(
        "You are a summaries assistant."
    )

    processor = DocumentProcessor(s3_client=mock_s3, bucket="test-bucket")
    result = processor.get_prompt("summarizer", "v1")

    assert result == "You are a summaries assistant."
    mock_s3.get_object.assert_called_once_with(
        Bucket="test-bucket",
        Key="prompts/summarizer/v1/system.txt",
    )


def test_store_document():
    mock_s3 = MagicMock()

    processor = DocumentProcessor(s3_client=mock_s3, bucket="test-bucket")
    key = processor.store_document(
        "knowledge-base",
        "doc-001",
        {"title": "Test", "content": "Test content"},
    )

    assert key == "documents/knowledge-base/doc-001.json"
    mock_s3.put_object.assert_called_once()
    call_kwargs = mock_s3.put_object.call_args[1]
    assert call_kwargs["Bucket"] == "test-bucket"
    assert call_kwargs["Key"] == "documents/knowledge-base/doc-001.json"

    body = json.loads(call_kwargs["Body"].decode("utf-8"))
    assert body["title"] == "Test"


def test_save_response():
    mock_s3 = MagicMock()

    processor = DocumentProcessor(s3_client=mock_s3, bucket="test-bucket")
    key = processor.save_response("req-abc123", {"status": "ok"})

    assert key.startswith("responses/")
    assert "req-abc123" in key
    mock_s3.put_object.assert_called_once()

These tests run in milliseconds, without Docker, without LocalStack, without AWS. The MagicMock replaces the boto3 client, and DocumentProcessor works identically because it doesn't care who the client is — only that it has .get_object() and .put_object().

Integration tests with the real factory

"""tests/test_integration.py — Integration tests with the real factory."""

import pytest
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor


@pytest.fixture
def local_settings():
    return Settings(
        environment=EnvironmentName.LOCAL,
        aws_endpoint_url="http://localhost:4566",
        aws_access_key_id="test",
        aws_secret_access_key="test",
        s3_bucket="test-integration-bucket",
    )


@pytest.fixture
def processor(local_settings):
    factory = ClientFactory(local_settings)
    s3 = factory.s3
    try:
        s3.create_bucket(Bucket=local_settings.s3_bucket)
    except s3.exceptions.BucketAlreadyOwnedByYou:
        pass

    return DocumentProcessor(s3_client=s3, bucket=local_settings.s3_bucket)


def test_roundtrip_prompt(processor):
    """Uploads and retrieves a prompt template — real integration."""
    processor.s3.put_object(
        Bucket=processor.bucket,
        Key="prompts/test/v1/system.txt",
        Body=b"Test prompt content",
    )
    result = processor.get_prompt("test", "v1")
    assert result == "Test prompt content"


def test_roundtrip_document(processor):
    """Uploads and verifies a document — real integration."""
    key = processor.store_document(
        "test-collection",
        "doc-integration",
        {"title": "Integration Test", "status": "active"},
    )

    response = processor.s3.get_object(Bucket=processor.bucket, Key=key)
    import json
    data = json.loads(response["Body"].read().decode("utf-8"))
    assert data["title"] == "Integration Test"

Advanced Pattern: Client with Built-in Retry

A wrapper that adds retry to any client

"""clients/resilient.py — Client wrapper with automatic retry."""

import time
import logging
from typing import Any

logger = logging.getLogger(__name__)


class ResilientClient:
    """Wrapper that adds automatic retry to a boto3 client.

    It doesn't modify the original client — it wraps calls with retry logic.
    """

    def __init__(
        self,
        client: Any,
        max_retries: int = 3,
        base_delay: float = 1.0,
        service_name: str = "unknown",
    ):
        self._client = client
        self._max_retries = max_retries
        self._base_delay = base_delay
        self._service_name = service_name

    def __getattr__(self, name: str):
        attr = getattr(self._client, name)
        if not callable(attr):
            return attr

        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, self._max_retries + 1):
                try:
                    return attr(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < self._max_retries:
                        delay = self._base_delay * (2 ** (attempt - 1))
                        logger.warning(
                            f"[{self._service_name}] {name} failed (attempt {attempt}/"
                            f"{self._max_retries}): {e}. Retry in {delay}s"
                        )
                        time.sleep(delay)
                    else:
                        logger.error(
                            f"[{self._service_name}] {name} failed after "
                            f"{self._max_retries} attempts: {e}"
                        )
            raise last_exception

        return wrapper

Factory with resilient clients

"""clients/factory.py — Extension with support for resilient clients."""

from clients.resilient import ResilientClient


class ClientFactory:
    # ... (previous code) ...

    def get_resilient_client(self, service: str) -> ResilientClient:
        """Returns a client with automatic retry."""
        base_client = self.get_client(service)
        return ResilientClient(
            client=base_client,
            max_retries=self.settings.max_retries,
            base_delay=1.0,
            service_name=service,
        )

    @property
    def resilient_s3(self) -> ResilientClient:
        return self.get_resilient_client("s3")

Usage:

factory = ClientFactory(settings)
s3 = factory.resilient_s3

# This automatically retries if S3 fails temporarily
s3.get_object(Bucket="my-bucket", Key="my-key")

Troubleshooting

Problem 1: "AttributeError: 'MagicMock' has no attribute 'exceptions'"

When mocking S3 for tests, s3.exceptions.ClientError fails because MagicMock doesn't have exceptions.

from botocore.exceptions import ClientError

# Instead of mock_s3.exceptions.ClientError, import it directly:
mock_s3.head_bucket.side_effect = ClientError(
    {"Error": {"Code": "404", "Message": "Not Found"}},
    "HeadBucket",
)

Problem 2: The factory caches a client that later fails

If LocalStack restarts, the cached client loses its connection.

class ClientFactory:
    def reset_client(self, service: str):
        """Invalidates the cache of a specific client."""
        self._clients.pop(service, None)

    def reset_all(self):
        """Invalidates all cached clients."""
        self._clients.clear()

Problem 3: Circular import between settings and factory

settings.py imports from factory.py which imports from settings.py.

# Solution: the factory imports Settings, not the loader
# factory.py
from config.settings import Settings  # ← the class, not the loader

# app.py (wiring)
from config.loader import get_settings
from clients.factory import ClientFactory

settings = get_settings()
factory = ClientFactory(settings)

Problem 4: The Lambda handler needs the container but can't import config

The handler must work as an entry point without circular dependencies.

# handler.py — lazy initialization
container = None

def get_container():
    global container
    if container is None:
        from services.container import ServiceContainer
        container = ServiceContainer.create()
    return container

Practical Exercises

Exercise 1: Factory with conditional SageMaker

Extend the ClientFactory so it only creates the SageMaker client if feature_sagemaker_enabled=True in settings. If the SageMaker client is requested when it isn't enabled, raise a descriptive exception.

import boto3
from typing import Any
from config.settings import Settings


class SageMakerDisabledError(Exception):
    pass


class ClientFactory:
    def __init__(self, settings: Settings):
        self.settings = settings
        self._base_kwargs = self._build_base_kwargs()
        self._clients: dict[str, Any] = {}

    def _build_base_kwargs(self) -> dict:
        kwargs = {"region_name": self.settings.aws_region}
        if self.settings.aws_endpoint_url:
            kwargs["endpoint_url"] = self.settings.aws_endpoint_url
        if self.settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = self.settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = self.settings.aws_secret_access_key
        return kwargs

    def get_client(self, service: str) -> Any:
        if service not in self._clients:
            self._clients[service] = boto3.client(service, **self._base_kwargs)
        return self._clients[service]

    @property
    def s3(self) -> Any:
        return self.get_client("s3")

    @property
    def sagemaker(self) -> Any:
        if not self.settings.feature_sagemaker_enabled:
            raise SageMakerDisabledError(
                f"SageMaker is not enabled in environment "
                f"'{self.settings.environment}'. "
                f"Set FEATURE_SAGEMAKER_ENABLED=true to enable it."
            )
        return self.get_client("sagemaker-runtime")


# Test
from config.settings import Settings, EnvironmentName

local = Settings(environment=EnvironmentName.LOCAL, feature_sagemaker_enabled=False)
factory = ClientFactory(local)

# S3 works
print(f"S3 client: {factory.s3}")

# SageMaker raises an error
try:
    factory.sagemaker
except SageMakerDisabledError as e:
    print(f"✅ Expected: {e}")

# Enabled in staging
staging = Settings(
    environment=EnvironmentName.STAGING,
    feature_sagemaker_enabled=True,
)
factory_staging = ClientFactory(staging)
print(f"SageMaker client: {factory_staging.sagemaker}")

Exercise 2: Service registry pattern

Implement a ServiceRegistry where services are registered with a name and resolved by that name. Useful when you have multiple implementations of the same service.

from typing import Any, Callable


class ServiceNotFoundError(Exception):
    pass


class ServiceRegistry:
    """Registry where services are registered and resolved by name."""

    def __init__(self):
        self._factories: dict[str, Callable] = {}
        self._instances: dict[str, Any] = {}

    def register(self, name: str, factory: Callable):
        """Registers a factory function for a service."""
        self._factories[name] = factory

    def resolve(self, name: str) -> Any:
        """Resolves a service by name (lazy, singleton)."""
        if name not in self._instances:
            if name not in self._factories:
                raise ServiceNotFoundError(
                    f"Service '{name}' not registered. "
                    f"Available: {list(self._factories.keys())}"
                )
            self._instances[name] = self._factories[name]()
        return self._instances[name]

    def reset(self, name: str | None = None):
        """Resets cached instances."""
        if name:
            self._instances.pop(name, None)
        else:
            self._instances.clear()

    @property
    def registered(self) -> list[str]:
        return list(self._factories.keys())


# Usage
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor

settings = Settings(environment=EnvironmentName.LOCAL)
factory = ClientFactory(settings)

registry = ServiceRegistry()

registry.register(
    "processor",
    lambda: DocumentProcessor(s3_client=factory.s3, bucket=settings.s3_bucket),
)

registry.register(
    "inference",
    lambda: InferenceService(
        lambda_client=factory.lambda_client,
        function_name=settings.lambda_function_name,
    ),
)

print(f"Registered services: {registry.registered}")

processor = registry.resolve("processor")
print(f"Processor: {processor}")

# Resolving again returns the same instance
processor2 = registry.resolve("processor")
assert processor is processor2
print("✅ Singleton verified")

Exercise 3: Client pool for parallel operations

Create a ClientPool that maintains multiple instances of an S3 client to allow concurrent operations (useful for bulk uploads).

import boto3
from typing import Any
from queue import Queue
from contextlib import contextmanager
from config.settings import Settings


class ClientPool:
    """Pool of boto3 clients for concurrent operations."""

    def __init__(self, settings: Settings, service: str, pool_size: int = 5):
        self.settings = settings
        self.service = service
        self._pool: Queue = Queue(maxsize=pool_size)

        kwargs = {"region_name": settings.aws_region}
        if settings.aws_endpoint_url:
            kwargs["endpoint_url"] = settings.aws_endpoint_url
        if settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

        for _ in range(pool_size):
            self._pool.put(boto3.client(service, **kwargs))

    @contextmanager
    def get_client(self):
        """Gets a client from the pool (context manager)."""
        client = self._pool.get()
        try:
            yield client
        finally:
            self._pool.put(client)

    @property
    def available(self) -> int:
        return self._pool.qsize()


# Usage
from concurrent.futures import ThreadPoolExecutor
import json

settings = Settings(
    environment="local",
    aws_endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    s3_bucket="pool-test",
)

pool = ClientPool(settings, "s3", pool_size=3)
print(f"Available clients: {pool.available}")


def upload_document(doc_id: int):
    with pool.get_client() as s3:
        s3.put_object(
            Bucket=settings.s3_bucket,
            Key=f"documents/doc-{doc_id:04d}.json",
            Body=json.dumps({"id": doc_id, "data": f"Document {doc_id}"}).encode(),
        )
    return doc_id


# Parallel upload using the pool
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(upload_document, range(10)))

print(f"Documents uploaded: {len(results)}")
print(f"Available clients after: {pool.available}")

Exercise 4: Factory with a built-in circuit breaker

Extend ClientFactory so that health_check() activates a circuit breaker if a service fails repeatedly, returning a "noop" client that does nothing instead of a real client.

import time
import boto3
from typing import Any
from config.settings import Settings


class NoopClient:
    """Client that does nothing — returns empty responses."""

    def __init__(self, service: str, reason: str):
        self._service = service
        self._reason = reason

    def __getattr__(self, name):
        def noop(*args, **kwargs):
            raise RuntimeError(
                f"Service '{self._service}' disabled: {self._reason}"
            )
        return noop


class CircuitBreakerFactory:
    """Factory with a per-service circuit breaker."""

    def __init__(self, settings: Settings, failure_threshold: int = 3, reset_timeout: int = 60):
        self.settings = settings
        self._failure_threshold = failure_threshold
        self._reset_timeout = reset_timeout
        self._failures: dict[str, int] = {}
        self._last_failure: dict[str, float] = {}
        self._clients: dict[str, Any] = {}

        self._base_kwargs = {"region_name": settings.aws_region}
        if settings.aws_endpoint_url:
            self._base_kwargs["endpoint_url"] = settings.aws_endpoint_url
        if settings.aws_access_key_id:
            self._base_kwargs["aws_access_key_id"] = settings.aws_access_key_id
            self._base_kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    def get_client(self, service: str) -> Any:
        failures = self._failures.get(service, 0)
        last_fail = self._last_failure.get(service, 0)

        if failures >= self._failure_threshold:
            elapsed = time.time() - last_fail
            if elapsed < self._reset_timeout:
                return NoopClient(
                    service,
                    f"Circuit open: {failures} failures, "
                    f"reset in {self._reset_timeout - elapsed:.0f}s",
                )
            self._failures[service] = 0

        if service not in self._clients:
            self._clients[service] = boto3.client(service, **self._base_kwargs)
        return self._clients[service]

    def record_failure(self, service: str):
        self._failures[service] = self._failures.get(service, 0) + 1
        self._last_failure[service] = time.time()

    def record_success(self, service: str):
        self._failures[service] = 0

    def circuit_status(self) -> dict:
        status = {}
        for service in set(list(self._clients.keys()) + list(self._failures.keys())):
            failures = self._failures.get(service, 0)
            if failures >= self._failure_threshold:
                elapsed = time.time() - self._last_failure.get(service, 0)
                status[service] = {
                    "state": "open",
                    "failures": failures,
                    "reset_in": max(0, self._reset_timeout - elapsed),
                }
            elif failures > 0:
                status[service] = {"state": "half-open", "failures": failures}
            else:
                status[service] = {"state": "closed", "failures": 0}
        return status


# Demo
settings = Settings(environment="local")
factory = CircuitBreakerFactory(settings, failure_threshold=2, reset_timeout=10)

s3 = factory.get_client("s3")
print(f"S3 client: {type(s3).__name__}")

factory.record_failure("s3")
factory.record_failure("s3")

s3_after = factory.get_client("s3")
print(f"S3 after 2 failures: {type(s3_after).__name__}")

print(f"Circuit status: {factory.circuit_status()}")

Summary

• Dependency injection decouples your business logic from the infrastructure. DocumentProcessor doesn't know whether it talks to LocalStack, AWS, or a mock. It receives a client and works.
• The factory pattern centralizes client construction. A single place knows how to create boto3 clients for the current environment. The rest of the app just asks for clients.
• DI makes your tests fast and reliable. With mock clients, you test business logic in milliseconds with no infrastructure.
• The ServiceContainer organizes the wiring. It's where config + factory + services connect. The Lambda handler only needs the container.
• Resilient clients add retry without changing business logic. The wrapper is transparent — your service doesn't know there's retry.
• In the next capsule, you'll use these patterns for multi-environment testing — the same test suite against LocalStack and AWS.

Additional Resources

1. Martin Fowler — Dependency Injection — The foundational article on DI
2. Factory Method Pattern — Factory pattern reference
3. python-dependency-injector — DI framework for Python
4. boto3 Client Reference — How clients are created in boto3
5. unittest.mock — Python Docs — Mocking in Python
6. Circuit Breaker Pattern — Pattern for resilience