SQLAlchemy RDS IAM login

Question

我正在尝试将我们的SQLAlchemy连接与我们的RDS Aurora集群连接起来。问题是，就在我们连接之前，我们需要通过boto3.当我们在应用程序启动时检索此密码时，我们会在 30 分钟后出现问题(从那时起SQLAlchemy开始刷新连接，但身份验证令牌不再有效(。我知道在create_engine方法(示例(中使用 creator 方法有一些解决方法，但是这使得解决方案数据库引擎特定(例如带有 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'])