SQLAlchemy RDS IAM 登录

Question

我正在尝试将我们的SQLAlchemy连接与我们的RDS Aurora集群连接起来。问题是，就在我们连接之前，我们需要通过boto3. 当我们在应用程序启动时检索此密码时，我们会在 30 分钟后出现问题（从那时起SQLAlchemy开始刷新连接但身份验证令牌不再有效）。我知道在create_engine方法（示例）中有使用创建者方法的解决方法，但是这使得解决方案数据库引擎特定于（例如带有 psycopg2 驱动程序的 postgres）。

我想要一个不包括驱动程序选择的解决方案，让SQLAlchemy我为我做这件事。但是，就在SQLAlchemy我想要执行此 boto 调用并更改 URL的连接之前。我尝试了不同的方法，使用插件或自定义池。然而，我尝试的所有这些方法都没有成功，因为我无法在连接之前影响 url（使用插件）或者我无法获取 url（使用自定义池）。

你们都知道如何解决这个问题而无需将代码绑定到特定的 DBMS 驱动程序。想听听你的意见

Answer 1

我能想到的最好的选择是，如果您有办法确定您需要哪个数据库并使用作为创建者发送的工厂函数，例如，如果您有一个环境变量，您可以执行以下操作：

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'])