使用 Boto3 进行 IAM 身份验证的 SQLAlchemy 可刷新凭证

Question

我使用 Boto3 生成的身份验证令牌通过 Sqlalchemy 连接到 Amazon RDS：

self.client = boto3.client("rds", region_name="eu-central-1")
self.token = self.client.generate_db_auth_token(
   self.endpoint, self.port, self.user
)

connection_string = (
   f"postgresql://{urlquote(self.user)}:"
   f"{urlquote(self.token)}@{self.endpoint}:{self.port}/dummy"
)

self.engine = db.create_engine(connection_string)

问题是提供的令牌仅在 15 分钟内有效（我想使用临时凭据！），而且我现在不知道如何告诉 SQLAlchemy 在旧令牌过期时自动创建新的身份验证令牌。

我找到了创建者参数，但是当令牌无效时似乎不会调用此函数：

def get_new_connection(self):
    self.token = self.client.generate_db_auth_token(
        self.endpoint, self.port, self.user
    )
    conn = psycopg2.connect(
        f"dbname='dummy' user='{self.user}' host='{self.endpoint}' password='{self.token}'"
    )
    return conn


 self.engine = db.create_engine(
 connection_string, creator=self.get_new_connection
 )

是否有内置函数，或者有更优雅的方法来解决这个问题？

Answer 1

根据SQLAlchemy 文档，使用易失性身份验证凭据的“正确”方法是利用事件系统：

生成动态身份验证令牌

DialectEvents.do_connect() 也是动态插入可能在引擎的生命周期内发生变化的身份验证令牌的理想方法。例如，如果令牌由 get_authentication_token() 生成并通过令牌参数传递给 DBAPI，则可以将其实现为：

from sqlalchemy import event

engine = create_engine("postgresql://user@hostname/dbname")

@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    cparams['token'] = get_authentication_token()

尽管有多种使用 SQLAlchemy 事件系统的方法，但我强烈建议您首先阅读有关如何使用事件的文档，并根据您的特定需求选择正确的实现/机制。

根据这些信息，只要引擎需要建立新连接，我就能够生成新的 RDS IAM 密码。