Secret Manager Integration: AWS Secrets Manager

AWS Secrets Manager is a fully managed service that encrypts secrets (database passwords, API keys, OAuth tokens) and stores them with automatic versioning, rotation, and audit logging. Instead of storing secrets in environment variables or configuration files, your Python application fetches secrets from Secrets Manager at runtime using the AWS SDK. This approach provides encryption at rest, automatic rotation, fine-grained IAM access control, and a complete audit trail—essential for regulated industries (finance, healthcare, government) and large teams managing hundreds of applications.

This article covers how to integrate Python applications with Secrets Manager using boto3, implement client-side caching to avoid excessive API calls, and handle secret rotation without downtime. A Secrets Manager secret is a key-value pair (or JSON object) stored encrypted by AWS and versioned automatically.

Installing and Configuring boto3

Install the AWS SDK:

pip install boto3

Configure AWS credentials. The SDK looks for credentials in this order:

1. Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
2. IAM role (if running on EC2, ECS, or Lambda)
3. ~/.aws/credentials file
4. ~/.aws/config file

For local development:

aws configure
# Prompts for Access Key ID, Secret Access Key, region

For production, use IAM roles (no credentials in code).

Fetching Secrets from AWS Secrets Manager

Create a helper module to fetch secrets:

import json
import boto3
from botocore.exceptions import ClientError

def get_secret(secret_name: str, region: str = "us-east-1") -> dict:
    """
    Fetch a secret from AWS Secrets Manager.
    
    Args:
        secret_name: Name of the secret (e.g., "prod/database/postgres")
        region: AWS region
    
    Returns:
        Dictionary with secret key-value pairs
    
    Raises:
        ClientError: If secret is not found or access is denied
    """
    client = boto3.client("secretsmanager", region_name=region)
    
    try:
        response = client.get_secret_value(SecretId=secret_name)
    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        if error_code == "ResourceNotFoundException":
            raise ValueError(f"Secret '{secret_name}' not found in Secrets Manager")
        elif error_code == "AccessDeniedException":
            raise PermissionError(
                f"Access denied to secret '{secret_name}'. Check IAM permissions."
            )
        else:
            raise
    
    # Secrets Manager returns either 'SecretString' (JSON) or 'SecretBinary'
    if "SecretString" in response:
        return json.loads(response["SecretString"])
    else:
        # Binary secret (e.g., certificate); return as string
        return {"value": response["SecretBinary"].decode("utf-8")}

# Usage at application startup
try:
    db_secret = get_secret("prod/database/postgres")
    db_password = db_secret["password"]
    db_host = db_secret["host"]
except Exception as e:
    print(f"Failed to load secrets: {e}")
    exit(1)

In Secrets Manager, create a secret with JSON content:

{
  "host": "prod-db.internal",
  "port": 5432,
  "user": "postgres",
  "password": "super_secret_password_123"
}

Caching Secrets to Avoid API Throttling

Fetching a secret from Secrets Manager takes ~100–200 ms. Calling it on every request is slow and risks hitting AWS API rate limits. Implement client-side caching with a TTL:

import json
import boto3
import time
from functools import wraps
from typing import Dict, Tuple

class SecretCache:
    """Cache secrets in memory with TTL."""
    
    def __init__(self, ttl_seconds: int = 3600):
        self.ttl = ttl_seconds
        self._cache: Dict[str, Tuple[dict, float]] = {}
        self.client = boto3.client("secretsmanager")
    
    def get(self, secret_name: str) -> dict:
        """
        Get secret from cache or fetch from Secrets Manager.
        
        Returns cached secret if within TTL, otherwise fetches fresh.
        """
        now = time.time()
        
        # Check cache
        if secret_name in self._cache:
            secret_data, timestamp = self._cache[secret_name]
            if now - timestamp < self.ttl:
                return secret_data
        
        # Fetch fresh
        try:
            response = self.client.get_secret_value(SecretId=secret_name)
            secret_data = json.loads(response["SecretString"])
            self._cache[secret_name] = (secret_data, now)
            return secret_data
        except Exception as e:
            raise RuntimeError(f"Failed to fetch secret '{secret_name}': {e}")
    
    def invalidate(self, secret_name: str = None):
        """Clear cache for a secret (or all if secret_name is None)."""
        if secret_name:
            self._cache.pop(secret_name, None)
        else:
            self._cache.clear()

# Global cache (initialized once)
secret_cache = SecretCache(ttl_seconds=3600)

# Usage
db_secret = secret_cache.get("prod/database/postgres")

A 1-hour TTL balances freshness and performance: secrets are cached locally, avoiding constant API calls, but rotations take effect within an hour.

Integrating with Pydantic Settings

Combine Secrets Manager with Pydantic for type validation:

from pydantic import Field, field_validator
from pydantic_settings import BaseSettings
from typing import Optional
from secret_cache import secret_cache

class Settings(BaseSettings):
    database_host: str = Field(default="localhost")
    database_port: int = Field(default=5432)
    database_user: str = Field(default="postgres")
    database_password: str = Field(default="")
    
    api_key: str = Field(default="")
    aws_region: str = Field(default="us-east-1")
    
    @field_validator("database_password", mode="before")
    @classmethod
    def load_db_password(cls, v):
        # If env var is set, use it; otherwise fetch from Secrets Manager
        if v:
            return v
        try:
            secret = secret_cache.get("prod/database/postgres")
            return secret.get("password", "")
        except Exception as e:
            print(f"Warning: could not fetch password from Secrets Manager: {e}")
            return ""
    
    @field_validator("api_key", mode="before")
    @classmethod
    def load_api_key(cls, v):
        if v:
            return v
        try:
            secret = secret_cache.get("prod/api/external-service")
            return secret.get("api_key", "")
        except Exception:
            return ""

settings = Settings()

This pattern allows environment variables to override Secrets Manager, enabling flexibility for testing and local development.

Handling Secret Rotation Events

AWS Secrets Manager can automatically rotate secrets on a schedule. Your application must handle rotation events gracefully—closing old connections and reconnecting with new credentials. Use event subscriptions:

import json
import boto3

def lambda_handler(event, context):
    """
    AWS Lambda function triggered by Secrets Manager rotation event.
    For example, MySQL rotation: create new password, test connection, mark complete.
    """
    service_client = boto3.client("secretsmanager")
    
    secret = event["SecretId"]
    client_request_token = event["ClientRequestToken"]
    step = event["Step"]
    secret_version_id = event["SecretVersionId"]
    metadata = event["SecretVersionStages"]
    
    # The secret is rotated through these steps
    if step == "create":
        # Generate new secret (new password)
        create_secret(service_client, secret, client_request_token, secret_version_id)
    
    elif step == "set":
        # Set the new password in the target service (e.g., database)
        set_secret(service_client, secret, client_request_token, secret_version_id)
    
    elif step == "test":
        # Test that the new password works
        test_secret(service_client, secret, client_request_token, secret_version_id)
    
    elif step == "finish":
        # Mark rotation as complete
        finish_secret(service_client, secret, client_request_token, secret_version_id)
    
    else:
        raise ValueError(f"Invalid step parameter: {step}")

def set_secret(client, secret, token, version):
    """Connect to database and set new password."""
    # Fetch the new secret version
    response = client.get_secret_value(
        SecretId=secret,
        VersionId=version,
        VersionStage="AWSPENDING"
    )
    new_secret = json.loads(response["SecretString"])
    
    # Connect to database and set password
    import psycopg2
    conn = psycopg2.connect(
        host=new_secret["host"],
        user=new_secret["user"],
        password=new_secret["previous_password"]  # Use old password to connect
    )
    cursor = conn.cursor()
    cursor.execute(
        f"ALTER USER {new_secret['user']} WITH PASSWORD %s",
        (new_secret["password"],)
    )
    conn.commit()
    cursor.close()
    conn.close()

Your application must close stale database connections and re-fetch secrets when rotation occurs. Use a connection pool with a TTL:

import psycopg2.pool
import time

class RotationAwarePool:
    def __init__(self, secret_name: str, max_ttl: int = 300):
        self.secret_name = secret_name
        self.max_ttl = max_ttl
        self.pool = None
        self.last_rotation = time.time()
    
    def get_connection(self):
        # Refresh pool every max_ttl seconds
        if time.time() - self.last_rotation > self.max_ttl:
            if self.pool:
                self.pool.closeall()
            secret = secret_cache.get(self.secret_name)
            self.pool = psycopg2.pool.SimpleConnectionPool(
                1, 5,
                host=secret["host"],
                user=secret["user"],
                password=secret["password"]
            )
            self.last_rotation = time.time()
        
        return self.pool.getconn()

db_pool = RotationAwarePool("prod/database/postgres")
conn = db_pool.get_connection()

Secrets Manager vs Environment Variables

Aspect	Environment Variables	Secrets Manager
Encryption	None (plaintext)	Yes (KMS at rest)
Rotation	Manual (redeploy)	Automatic (no downtime)
Audit	No	Full audit log
Access Control	OS-level	Fine-grained IAM
Cost	Free	Pay per secret ($0.40/month per secret)
Best For	Local dev, non-sensitive config	Production, sensitive secrets, regulated environments

Use environment variables for local development and non-sensitive settings; use Secrets Manager for production credentials.

Key Takeaways

• AWS Secrets Manager encrypts secrets, provides automatic rotation, and logs all access—ideal for production environments.
• Fetch secrets using boto3.client("secretsmanager").get_secret_value() and cache them locally to avoid API throttling.
• Use IAM roles for access control, never embed access keys in application code.
• Implement connection pool refreshes with a TTL to handle secret rotation without downtime.
• Integrate Secrets Manager with Pydantic Settings to validate secret values and provide type safety.

Frequently Asked Questions

How do I rotate secrets without downtime?

Use AWS Secrets Manager's automatic rotation feature, which creates a new version and triggers Lambda functions to update the target service (database, API, etc.). Your application fetches secrets with a TTL, so new credentials are picked up within the cache window (typically 1 hour). Ensure your database connections refresh periodically.

What if Secrets Manager is down or unreachable?

Your application should fall back gracefully: cache secrets aggressively (12-hour TTL), log errors, and use environment variable overrides. In Lambda, where cold starts matter, pre-fetch secrets before handling requests.

Can I store complex secrets (like TLS certificates) in Secrets Manager?

Yes. Secrets Manager supports binary secrets (files) and JSON. For certificates, store as base64-encoded JSON: {"cert": "base64_encoded_pem", "key": "base64_encoded_key"}, then decode at runtime: base64.b64decode(secret["cert"]).

How do I test my application with Secrets Manager locally?

Use LocalStack (emulates AWS services locally) or mock boto3.client() in tests: from unittest.mock import MagicMock; mock_client.get_secret_value.return_value = {...}. For development, use a .env file that Pydantic reads before attempting Secrets Manager.

Does AWS charge per secret or per API call?

AWS charges $0.40 per secret per month (baseline) plus $0.05 per 10,000 API requests. For a typical application with 5 secrets and hourly cache refreshes, expect $2–5/month. Increase cache TTL to reduce API calls.

Further Reading

• AWS Secrets Manager User Guide
• boto3 Secrets Manager Documentation
• Secrets Manager Rotation Patterns
• LocalStack for Local Secret Manager Testing