🚀 [Solution] Context Manager for Using Specific Boto3 Sessions with PynamoDB Models #1280
Description
🚀 Ultimate Solution: PynamoDB with Specific Boto3 Sessions
Hi folks,
As the author of boto-session-manager and pynamodb-mate, I'm excited to share a comprehensive solution that addresses the long-standing need for using specific boto3 sessions with PynamoDB. (THIS SOLUTION HAS A PYTHON LIBRARY pynamodb_session_manager)
🎯 What This Solves
This solution enables you to:
- ✅ Use any specific boto3 session with PynamoDB models
- ✅ Switch between different AWS profiles/credentials seamlessly
- ✅ Declare your ORM models only once (no need for multiple model classes)
- ✅ Perform operations across multiple AWS accounts without side effects
- ✅ Maintain clean, readable code with context managers
📦 The Solution: this_boto_session_manager Context Manager
Here's the reusable context manager that makes it all possible:
# -*- coding: utf-8 -*-
"""
PynamoDB Context Manager for Using Specific Boto3 Sessions
This module provides a context manager that allows PynamoDB models to temporarily
use different AWS credentials/sessions without modifying the model definition.
This is particularly useful for:
- Multi-account operations
- Different AWS profiles for different environments
- Temporary credential switching
- Cross-account DynamoDB operations
Dependencies:
- pynamodb
- `boto-session-manager <https://pypi.org/project/boto-session-manager/>`_
Author: Sanhe Hu
Date: 2026-06-26
"""
import typing as T
import contextlib
from pynamodb.connection import Connection
if T.TYPE_CHECKING: # pragma: no cover
from boto_session_manager import BotoSesManager
from pynamodb.models import Model
@contextlib.contextmanager
def this_boto_session_manager(
bsm: "BotoSesManager",
table: T.Type["Model"],
):
"""
Context manager to temporarily switch PynamoDB model to use different AWS credentials.
This context manager allows a PynamoDB model to temporarily use different AWS
credentials by leveraging the boto-session-manager's awscli() context manager
and manipulating the model's connection and region settings.
Args:
bsm: The boto session manager instance containing the
target AWS credentials/profile to use
table: The PynamoDB model class that will use the new credentials
Yields:
None: This is a context manager that doesn't return a value
Usage::
# Define your model
class MyModel(Model):
class Meta:
table_name = "my_table"
region = "us-east-1"
id = NumberAttribute(hash_key=True)
# Use different credentials temporarily
target_bsm = BotoSesManager(profile_name="target_profile")
with target_aws_cred(target_bsm, MyModel):
# All operations here use the target_profile credentials
MyModel.create_table()
item = MyModel(id=1)
item.save()
# Back to original credentials
items = MyModel.scan() # This uses default credentials
How it works:
1. Saves the current region setting from the model's Meta class
2. Enters the boto session manager's awscli() context (sets env vars)
3. Clears the model's cached connection to force recreation
4. Updates the model's region to match the session manager's region
5. Creates a new Connection object with the new region
6. On exit, restores the original region and clears connection again
.. note::
- The model's _connection attribute is set to None to force PynamoDB
to create a new connection with the current environment variables
- The region is temporarily changed to ensure consistency with the session
- All changes are reverted when exiting the context manager
"""
# Store the current region setting to restore it later
current_aws_region = table.Meta.region
try:
# Enter the boto session manager's context
# This typically sets AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, etc. as env vars
with bsm.awscli():
# Clear the existing connection to force PynamoDB to create a new one
# with the current environment variables (which are now set by bsm.awscli())
table._connection = None
# Update the model's region to match the session manager's region
# This ensures consistency between the session and the model configuration
table.Meta.region = bsm.aws_region
# Create a new connection object with the new region
# This connection will inherit the AWS credentials from environment variables
Connection(region=bsm.aws_region)
# Yield control back to the caller
# All PynamoDB operations within this context will use the new credentials
yield None
finally:
# Cleanup: Restore the original state regardless of success or failure
# Clear the connection again to ensure clean state
table._connection = None
# Restore the original region setting
table.Meta.region = current_aws_region
# Create a new connection with the original region
# When the bsm.awscli() context exits, the environment variables
# will be restored, so this connection will use the original credentials
Connection(region=current_aws_region)💡 Usage Example
This example demonstrates switching between your default AWS profile and a project-specific profile:
# -*- coding: utf-8 -*-
from pynamodb_session_manager import this_boto_session_manager
from boto_session_manager import BotoSesManager
from pynamodb.models import Model
from pynamodb.attributes import NumberAttribute
from pynamodb.constants import PAY_PER_REQUEST_BILLING_MODE
# Configuration constants
aws_region = "us-east-1"
# Initialize session managers for different AWS accounts/profiles
# THE default AWS PROFILE on your machine (usually from ~/.aws/credentials [default])
# By default, PynamoDB will use this profile
default_bsm = BotoSesManager(region_name=aws_region)
print(f"default aws account: {default_bsm.aws_account_id = }")
# THE TARGET AWS PROFILE YOU WANT TO USE (specific profile from ~/.aws/credentials)
project_bsm = BotoSesManager(
profile_name="bmt_app_dev_us_east_1", region_name=aws_region
)
print(f"project aws account: {project_bsm.aws_account_id = }")
class User(Model):
"""
Example PynamoDB model for demonstration purposes.
This model represents a simple user table with an ID as the primary key.
It's configured to use pay-per-request billing mode for cost optimization
during testing.
"""
class Meta:
table_name = "pynamodb_connection_using_specific_boto3_session_test"
region = aws_region
billing_mode = PAY_PER_REQUEST_BILLING_MODE
id: int = NumberAttribute(hash_key=True)
# Example 1: Create table in the target account
print("\n=== Creating table in target account ===")
with this_boto_session_manager(project_bsm, User):
# This will create the table in the project AWS account
# because we're using project_bsm credentials within the context
User.create_table(wait=True)
# Example 2: Attempt operation with default credentials (should fail)
print("\n=== Attempting operation with default credentials ===")
# This will use the default AWS account credentials
# Since the table doesn't exist in the default account, it will raise an exception
try:
user = User(id=1)
user.save()
print("Unexpected success - table exists in default account")
except Exception as e:
print(f"Expected error (table not in default account): {e}")
# Example 3: Insert item in target account
print("\n=== Inserting item in target account ===")
with this_boto_session_manager(project_bsm, User):
# This will insert item to the table in the project AWS account
# because we're using project_bsm credentials within the context
user = User(id=1)
user.save()
print(f"User with ID=1 saved to account: {project_bsm.aws_account_id}")
# Example 4: Retrieve item from target account
print("\n=== Retrieving item from target account ===")
with this_boto_session_manager(project_bsm, User):
# This will retrieve the item from the table in the project AWS account
user = User.get(1)
print(
f"Retrieved user from account {project_bsm.aws_account_id}: {user.attribute_values}"
)
print("\n=== Demo completed ===")Example output
default aws account: default_bsm.aws_account_id = '111111111111'
project aws account: project_bsm.aws_account_id = '222222222222'
=== Creating table in target account ===
=== Attempting operation with default credentials ===
Expected error (table not in default account): Failed to put item: An error occurred (ResourceNotFoundException) on request (ABCDEFG) on table (pynamodb_connection_using_specific_boto3_session_test) when calling the PutItem operation: Requested resource not found
=== Inserting item in target account ===
User with ID=1 saved to account: 222222222222
=== Retrieving item from target account ===
Retrieved user from account 222222222222: {'id': 1}
=== Demo completed ===🔧 Key Benefits
- No Model Duplication: Define your PynamoDB models once, use them with any AWS credentials
- Thread-Safe: Each context properly manages connections and state
- Clean Restoration: Automatically restores original settings when exiting context
- Error-Safe: Uses try/finally to ensure cleanup even if exceptions occur
- Cross-Account Operations: Perfect for multi-tenant or multi-account architectures
Note: This is a workaround that leverages PynamoDB's internal connection management. While it works reliably, the ideal long-term solution would be native session support in PynamoDB itself.
Feel free to test this solution and provide feedback!