Back to Blogging Link to heading

It’s been a while since my last post. The combination of spring break, family illnesses, and an increasingly demanding work schedule forced me to break my weekly blogging cadence. But I’m finally getting back into the groove, and what better way to return than with a project that addresses a common pain point I’ve encountered multiple times throughout my career.

The RDS IAM Authentication Problem Link to heading

Like many developers working with AWS, I’ve repeatedly had to implement RDS IAM authentication for various work projects. It’s a common requirement, yet surprisingly, I’ve never found an out-of-the-box solution in any language that just handles it seamlessly. This frustration led me to finally address the issue head-on this weekend, by creating a Python library to simplify this process.

Embracing Python (Reluctantly) Link to heading

I’ll admit it - I’ve been skeptical of Python for most of my career. Much of my bias stems from brief interactions over 20 years ago and my overall familiarity with Ruby. The vast majority of my professional experience has been with Ruby, Java, and Go. Python just never filled a niche that I needed, and my most significant Python experience was a quarter-long greenfield project that got shut down before it could really take off.

Despite these reservations, I’m doing my best to adopt and be productive with Python. Creating this library was part of that effort - turning my recent experience into something that might provide value to others in the future.

Introducing sqlalchemy-rds-iam Link to heading

I’m pleased to introduce sqlalchemy-rds-iam, a Python library that provides a straightforward way to integrate RDS IAM authentication with SQLAlchemy. The interface is clean and simple:

from sqlalchemy import create_engine
from sqlalchemy_rds_iam import RDSIAMAuth

# Create the auth handler
auth = RDSIAMAuth()

# Create an engine
engine = create_engine(
    "postgresql+psycopg2://[email protected]:5432/dbname"
)

# Register the auth handler with the engine
auth.register_for_engine(engine)

# Use the engine as normal - IAM tokens will be automatically generated and renewed
with engine.connect() as conn:
    result = conn.execute("SELECT 1")

You can also customize the authentication with specific AWS credentials:

import boto3

session = boto3.Session(profile_name='custom-profile')
auth = RDSIAMAuth(boto_session=session)
engine = create_engine(
    "postgresql+psycopg2://[email protected]:5432/dbname"
)
auth.register_for_engine(engine)

The library automatically handles token expiration during use - if your connection lives longer than the token’s validity period, the next query will transparently generate a fresh token without any action required from your code.

Deep Dives and Learnings Link to heading

Creating this library involved several interesting deep dives into SQLAlchemy and Python best practices:

Event-Driven Authentication Link to heading

At work, I had previously used a creator function with SQLAlchemy, but research showed this wasn’t the most current approach. Instead, the recommended pattern uses the engine event system with do_connect.

Rather than using the @listen_for decorator (which caused issues with nested helper methods and static type checking), I implemented direct event registration:

listen(engine, "do_connect", self.provide_token)

This method worked much better with tools like mypy and offered a cleaner integration with SQLAlchemy.

Efficient Cache Keys Link to heading

Another interesting learning came when implementing token caching. Initially, I used a formatted string as the cache key, but Sourcery.ai helpfully pointed out that this wasn’t the optimal approach. Using a tuple instead proved more efficient:

# Create cache key from input parameters using a tuple instead of a string
cache_key = (user, host, port, effective_region)

This small change improved both performance and code quality. Tuples are more memory-efficient than strings for caching because they don’t require string formatting operations and can directly use Python’s internal hashing of the composite values, making lookups faster and reducing memory overhead.

Thread Safety and Token Management Link to heading

One aspect I’m particularly proud of is the thread-safe token management system. The library automatically:

  1. Generates tokens when needed
  2. Caches them for efficiency (configurable timeout, defaults to 10 minutes)
  3. Handles regeneration when they expire
  4. Does all of this in a thread-safe manner

The implementation uses a combination of threading.Lock and TTLCache from the cachetools library:

self._token_cache: TTLCache[Tuple[str, str, int, str], str] = TTLCache(
    maxsize=100, ttl=cache_timeout
)
self._cache_lock = threading.Lock()

Still Skeptical, But Coming Around Link to heading

I’m still somewhat skeptical of Python, but I must admit that the available tooling makes it much easier to work with than it was when I first encountered it. Tools like mypy, black, isort, and ruff have significantly improved the development experience, making Python feel more structured and predictable.

Creating this library has been a valuable experience in working with Python in a more structured way. While I may not be a Python convert just yet, I appreciate the ecosystem more than I did before.

Looking Forward Link to heading

If you find yourself needing to connect to RDS using IAM authentication with SQLAlchemy, give sqlalchemy-rds-iam a try. It’s available on PyPI and supports Python 3.9 to 3.13. Contributions are welcome!

For detailed information, check out the GitHub repository which includes comprehensive documentation and examples.

I’ll be continuing my Python journey and hope to share more learnings in future posts. My next steps for the library include adding more database drivers beyond PostgreSQL and potentially exploring integration with other AWS services.

One significant improvement I’m planning to implement is background token refreshing. Currently, token generation happens in the “hot path” of execution - meaning it can potentially add latency to your database connections when tokens need refreshing. At work, I implemented a version that refreshes tokens in the background, keeping them ready before they’re needed. Since my work implementation was built in an async framework, I need to do more research into the best way to implement this in a general-purpose Python library that supports both synchronous and asynchronous code patterns.

If you’re using this library, I’d love to hear about your use cases and any suggestions for improvements.

Maybe I’ll even become a Python enthusiast one day - stranger things have happened in my decades-long programming career.