Skip to content

infrastructure.secrets.aws_adapter

src.infrastructure.secrets.aws_adapter

AWS Secrets Manager adapter for production secrets.

Implements SecretsProtocol using AWS Secrets Manager with in-memory caching.

File: aws_adapter.py → class AWSAdapter (PEP 8 naming)

Classes

AWSAdapter

Bases: BaseSecretsAdapter

Production secrets from AWS Secrets Manager.

Features
  • In-memory caching (reduce API calls, cost savings)
  • Hierarchical naming: /dashtam/{env}/{category}/{name}
  • Automatic retry with exponential backoff (boto3 default)
Caching Strategy
  • Cache secrets in memory to reduce AWS API calls
  • Cost: $0.05 per 10,000 API calls
  • Without cache: ~100,000 calls/month = $0.50
  • With cache: ~1,000 calls/month (startup only) = $0.005
  • Savings: 99% reduction in API costs
Source code in src/infrastructure/secrets/aws_adapter.py 
 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
class AWSAdapter(BaseSecretsAdapter):
    """Production secrets from AWS Secrets Manager.

    Features:
        - In-memory caching (reduce API calls, cost savings)
        - Hierarchical naming: /dashtam/{env}/{category}/{name}
        - Automatic retry with exponential backoff (boto3 default)

    Caching Strategy:
        - Cache secrets in memory to reduce AWS API calls
        - Cost: $0.05 per 10,000 API calls
        - Without cache: ~100,000 calls/month = $0.50
        - With cache: ~1,000 calls/month (startup only) = $0.005
        - Savings: 99% reduction in API costs
    """

    def __init__(self, environment: str, region: str = "us-east-1") -> None:
        """Initialize AWS Secrets Manager client.

        Args:
            environment: 'staging' or 'production'.
            region: AWS region for secrets (default: us-east-1).

        Raises:
            ImportError: If boto3 not installed.
        """
        try:
            import boto3
        except ImportError as e:
            raise ImportError(
                "boto3 required for AWS Secrets Manager. Install with: uv add boto3"
            ) from e

        self.client = boto3.client("secretsmanager", region_name=region)
        self.environment = environment
        self._cache: dict[str, str] = {}

    def get_secret(self, secret_path: str) -> Result[str, SecretsError]:
        """Get secret from AWS Secrets Manager.

        Args:
            secret_path: Path like 'database/url' or 'schwab/api_key'.

        Returns:
            Success(secret_value) if found in AWS.
            Failure(SecretsError) if not found or access denied.

        Example:
            >>> adapter = AWSAdapter(environment="production")
            >>> result = adapter.get_secret("database/url")
            >>> # Fetches from: /dashtam/production/database/url
            >>> # Success("postgresql://...")
        """
        secret_id = f"/dashtam/{self.environment}/{secret_path}"

        # Check cache first
        if secret_id in self._cache:
            return Success(value=self._cache[secret_id])

        try:
            response = self.client.get_secret_value(SecretId=secret_id)
            secret_value = response["SecretString"]

            # Cache for future calls
            self._cache[secret_id] = secret_value
            return Success(value=secret_value)

        except self.client.exceptions.ResourceNotFoundException:
            return Failure(
                error=SecretsError(
                    code=ErrorCode.SECRET_NOT_FOUND,
                    message=f"Secret not found in AWS: {secret_id}",
                )
            )
        except Exception as e:
            return Failure(
                error=SecretsError(
                    code=ErrorCode.SECRET_ACCESS_DENIED,
                    message=f"Failed to access AWS secret: {secret_id}",
                    details={"error": str(e)},
                )
            )

    # get_secret_json() inherited from BaseSecretsAdapter

    def refresh_cache(self) -> None:
        """Clear cache to force reload on next access.

        Call this after rotating secrets in AWS console or Terraform.
        Next get_secret() call will fetch fresh value from AWS.

        Example:
            >>> adapter = AWSAdapter(environment="production")
            >>> adapter.refresh_cache()  # Clear cache
            >>> result = adapter.get_secret("database/password")
            >>> # Fetches fresh from AWS
        """
        self._cache.clear()
Functions
__init__ 
__init__(
    environment: str, region: str = "us-east-1"
) -> None

Parameters:

Name Type Description Default
environment str

'staging' or 'production'.

 required
region str

AWS region for secrets (default: us-east-1).

 'us-east-1'

Raises:

Type Description
ImportError

If boto3 not installed.
Source code in src/infrastructure/secrets/aws_adapter.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self, environment: str, region: str = "us-east-1") -> None:
    """Initialize AWS Secrets Manager client.

    Args:
        environment: 'staging' or 'production'.
        region: AWS region for secrets (default: us-east-1).

    Raises:
        ImportError: If boto3 not installed.
    """
    try:
        import boto3
    except ImportError as e:
        raise ImportError(
            "boto3 required for AWS Secrets Manager. Install with: uv add boto3"
        ) from e

    self.client = boto3.client("secretsmanager", region_name=region)
    self.environment = environment
    self._cache: dict[str, str] = {}
get_secret 
get_secret(secret_path: str) -> Result[str, SecretsError]

Get secret from AWS Secrets Manager.

Parameters:

Name Type Description Default
secret_path str

Path like 'database/url' or 'schwab/api_key'.

 required

Returns:

Type Description
Result[str, SecretsError]

Success(secret_value) if found in AWS.
Result[str, SecretsError]

Failure(SecretsError) if not found or access denied.
Example

adapter = AWSAdapter(environment="production") result = adapter.get_secret("database/url")

Fetches from: /dashtam/production/database/url
Success("postgresql://...")
Source code in src/infrastructure/secrets/aws_adapter.py 
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
def get_secret(self, secret_path: str) -> Result[str, SecretsError]:
    """Get secret from AWS Secrets Manager.

    Args:
        secret_path: Path like 'database/url' or 'schwab/api_key'.

    Returns:
        Success(secret_value) if found in AWS.
        Failure(SecretsError) if not found or access denied.

    Example:
        >>> adapter = AWSAdapter(environment="production")
        >>> result = adapter.get_secret("database/url")
        >>> # Fetches from: /dashtam/production/database/url
        >>> # Success("postgresql://...")
    """
    secret_id = f"/dashtam/{self.environment}/{secret_path}"

    # Check cache first
    if secret_id in self._cache:
        return Success(value=self._cache[secret_id])

    try:
        response = self.client.get_secret_value(SecretId=secret_id)
        secret_value = response["SecretString"]

        # Cache for future calls
        self._cache[secret_id] = secret_value
        return Success(value=secret_value)

    except self.client.exceptions.ResourceNotFoundException:
        return Failure(
            error=SecretsError(
                code=ErrorCode.SECRET_NOT_FOUND,
                message=f"Secret not found in AWS: {secret_id}",
            )
        )
    except Exception as e:
        return Failure(
            error=SecretsError(
                code=ErrorCode.SECRET_ACCESS_DENIED,
                message=f"Failed to access AWS secret: {secret_id}",
                details={"error": str(e)},
            )
        )
refresh_cache 
refresh_cache() -> None

Clear cache to force reload on next access.

Call this after rotating secrets in AWS console or Terraform. Next get_secret() call will fetch fresh value from AWS.

Example

adapter = AWSAdapter(environment="production") adapter.refresh_cache() # Clear cache result = adapter.get_secret("database/password")

Fetches fresh from AWS
Source code in src/infrastructure/secrets/aws_adapter.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
def refresh_cache(self) -> None:
    """Clear cache to force reload on next access.

    Call this after rotating secrets in AWS console or Terraform.
    Next get_secret() call will fetch fresh value from AWS.

    Example:
        >>> adapter = AWSAdapter(environment="production")
        >>> adapter.refresh_cache()  # Clear cache
        >>> result = adapter.get_secret("database/password")
        >>> # Fetches fresh from AWS
    """
    self._cache.clear()