SQLAlchemy RDS IAM 登录

Question

我正在尝试连接我们的 SQLAlchemy与我们的RDS Aurora 联系簇。问题是，在我们连接之前，我们需要通过 boto3 获取身份验证 token 。 .当我们在应用程序启动时检索此密码一次时，我们会在 30 分钟后遇到问题(从那时起 SQLAlchemy 开始刷新连接，但身份验证 token 不再有效)。我知道在 create_engine 中有使用创建者方法的解决方法。方法(example)，但是这使得解决方案数据库引擎特定(例如带有 psycopg2 驱动程序的 postgres)。

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

你们都知道如何解决这个问题而无需将代码绑定(bind)到特定的 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'])