I have previously written about sharing database connections when using FastAPI and it’s something I come back to and reuse whenever using FastAPI and postgres (which is often). In fact, it seems to be the most read post on the blog.
Recently, I came to do this again with an app running in AWS where we wanted to use RDS IAM. RDS IAM can be really useful. By deferring to IAM to authenticate apps to the DB rather than using manually managed database users, stacks running with Infrastructure as Code are a lot easier to deploy. The choice is analogous to using SSM to authenticate to EC2 instances rather than managing SSH keys.
There’s a small wrinkle though, the RDS IAM credentials change every 15 minutes, so if you have a connection pool like in my other post and your app lasts longer than 15 minutes it would break (at least new connections will fail)! Thankfully a recent version of psycopg_pool (3.3.0) made this really easy to fix.
Setting up a basic connection pool
Let’s remind ourselves of how we integrated psycopg pool with a static connection string. We tie the opening of the pool (establishing connections) to FastAPI’s lifespan hook so we open the connections as FastAPI is starting up and we have an asyncio event loop (and make sure we close connections when shutting down).
In this example, conninfo is a typical postgres connection string.
import psycopg_pool
from contextlib import asynccontextmanager
from fastapi import FastAPI
conninfo = "postgresql://user:password@localhost:5432/dbname"
pool = psycopg_pool.AsyncConnectionPool(
conninfo=conninfo,
open=False,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
await pool.open()
yield
await pool.close()
app = FastAPI(lifespan=lifespan)Making the conninfo a callable
It looks like a few people have had similar requirements for rotating database passwords over the years, and psycopg_pool version 3.3.0 implemented passing conninfo as a Callable. Now, we can let new connections use different credentials all in the same connection pool. I don’t think we need to worry about already open connections as they’ve already authenticated to postgres.
from pydantic import SecretStr
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_host: str = "localhost"
database_port: int = 5432
database_dbname: str = "db"
database_username: str = "postgres"
database_password: SecretStr = SecretStr("")
def get_database_url(self) -> str:
host = self.database_host
port = self.database_port
dbname = self.database_dbname
user = self.database_username
passwd = self.database_password.get_secret_value()
return f"postgresql://{user}:{passwd}@{host}:{port}/{dbname}"
settings = Settings()
pool = psycopg_pool.AsyncConnectionPool(
conninfo=settings.get_database_url, # now a callable!
open=False,
)This is actually still static credentials, but now we can change get_database_url() to update credentials as and when we need. You might not be using RDS IAM but still want to have a rotating database password (or you’re using other cloud databases).
Generating RDS IAM auth tokens
When using RDS IAM auth, we generate database passwords using boto3 instead. It’s worth understanding that this isn’t making any external API calls; to generate a password, boto3 uses your credentials (AWS Access Key/Secret) and the current timestamp to produce a cryptographic signature. We send that to RDS and AWS’s IAM validates the signature. This signature is AWS’s SigV4 that is used across AWS for authenticating HTTP requests (e.g. when using S3). Slightly awkwardly for us, the SigV4 token is a URL, which we are including inside another URL, so we need to escape it.
from urllib.parse import quote_plus
import boto3
client = boto3.client("rds")
class Settings(BaseSettings):
# ...existing fields...
use_rds_iam_auth: bool = True
def get_database_url(self) -> str:
host = self.database_host
port = self.database_port
dbname = self.database_dbname
user = self.database_username
if self.use_rds_iam_auth:
token = client.generate_db_auth_token(DBHostname=host, Port=port, DBUsername=user)
passwd = quote_plus(token)
return f"postgresql://{user}:{passwd}@{host}:{port}/{dbname}?sslmode=require"
passwd = self.database_password.get_secret_value()
return f"postgresql://{user}:{passwd}@{host}:{port}/{dbname}"NB: you’ll need to
grant rds_iam to <user>in your RDS database before you can connect with RDS IAM.
This works nicely! By having a use_rds_iam_auth flag we can keep using ‘normal’ postgres authentication when developing locally.
This was fairly easy because of conninfo being now callable as of the recent version 3.3.0 so thanks to the psycopg maintainers for that update - before you had to subclass the connection pool.
Optional extra: working over a local port forward
This can become a little more awkward depending on how you are using RDS. I’d expect most people run their database in a private subnet and do not have it accessible on the internet, but there might be times when you want to connect to it with your app running locally for development purposes.
We all have a development database, some people are sensible enough that this is not also their production database :D
If it’s in a private subnet you’ll have a tunnel/port-forward to connect to it from a dev machine. In that case RDS IAM will not work straight away, if you are connecting to localhost. You’ll see above that to generate the RDS IAM token we need to supply the hostname, which will be <something>.rds.amazonaws.com. If we swap the hostname to localhost for our port forward, we will generate a SigV4 with localhost, which RDS will reject.
Instead we need to separate the hostname for signature from the connection hostname. I’ve added a pydantic validator here to make sure it isn’t inadvertently missed. At this point we might be over engineering but if you are regularly port forwarding to RDS then I think it’s worth having!
from ipaddress import ip_address
from pydantic import model_validator
from pydantic_settings import BaseSettings
LOCALHOST_HOSTS = ["localhost", "localhost.localdomain"]
def is_loopback(host: str) -> bool:
try:
return ip_address(host).is_loopback
except ValueError:
return host in LOCALHOST_HOSTS
client = boto3.client("rds")
class Settings(BaseSettings):
# existing fields...
rds_endpoint_for_signature: str = "" # "foo.rds.amazonaws.com"
@model_validator(mode="after")
def validate_rds_iam_config(self) -> "Settings":
if self.use_rds_iam_auth and is_loopback(self.database_host) and not self.rds_endpoint_for_signature:
raise ValueError(
"rds_endpoint_for_signature must be set when use_rds_iam_auth is True and database_host is loopback"
)
return self
def get_database_url(self) -> str:
host = self.database_host
port = self.database_port
dbname = self.database_dbname
user = self.database_username
if self.use_rds_iam_auth:
token_hostname = self.rds_endpoint_for_signature if is_loopback(host) else host
token = client.generate_db_auth_token(DBHostname=token_hostname, Port=port, DBUsername=user)
passwd = quote_plus(token)
return f"postgresql://{user}:{passwd}@{host}:{port}/{dbname}?sslmode=require"
passwd = self.database_password.get_secret_value()
return f"postgresql://{user}:{passwd}@{host}:{port}/{dbname}"