SQLAlchemy RDS IAM login

Question

I am trying to connect our SQLAlchemy connections with our RDS Aurora cluster. The issue is that just before we connect we need to get a auth token via the boto3. When we retrieve this password once on application startup we get issues after 30 minutes (since then SQLAlchemy starts refreshing the connections but the auth token is not valid anymore). I know there are workarounds using the creator method in the create_engine method (example), however this makes the solution database engine specific (for example postgres with psycopg2 driver).

I would like to have a solution that doesn't include a choice of driver and let SQLAlchemy do this for me. However, just before the connection of SQLAlchemy I want to do this boto call and change the URL. I have tried different approaches, using a plugin or a custom Pool. However all those approaches I tried didn't succeed due to the fact that I couldn't influence the url before connection (with a plugin) or that I couldn't get the url (using a custom pool).

Do you all have an idea how I could fix this issue without the need to bind the code to a specific DBMS driver. Would like to hear from you

Answer 1

The best option I can think of is if you have a way to determine which database you need and use a factory function send as creator for example, if you have an environment variable with that you could do something like this:

import os
from functools import partial

def connection_factory(database, *args, **kwargs):
    try:
        return partial(FACTORIES[database], *args, **kwargs)
    except ImportError as e:
        raise RuntimeError(f'No driver installed for database {database}' from e
    except IndexError as e:
        raise RuntimeError(f'No factory for database {database}')

def postgresql_factory(*args, **kwargs):
    import psycopg2
    return psycopg2.connect(..., *args, **kwargs)

def mysql_factory():
    import MySQLdb
    return MySQLdb.connect(..., *args, **kwarg)

FACTORIES = {'postgres': postgresql_factory, 'mysql': mysql_factory}

engine = create_engine(..., creator=creator_factory(os.environ['DATABASE_ENGINE'])