我正在使用 Boto3 生成的身份验证 token 通过 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)
问题是提供的 token 仅在 15 分钟内有效(我想使用临时凭据!),我现在不知道如何告诉 SQLAlchemy 在旧 token 过期时自动创建新的身份验证 token 。
我找到了 creator 参数,但是当 token 无效时似乎不会调用此函数:
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
)
是否有内置功能,或更优雅的方法来解决这个问题?
最佳答案
根据SQLAlchemy documentation ,使用 volatile 身份验证凭据的“正确”方法是利用事件系统:
Generating dynamic authentication tokens
DialectEvents.do_connect() is also an ideal way to dynamically insert an authentication token that might change over the lifespan of an Engine. For example, if the token gets generated by get_authentication_token() and passed to the DBAPI in a token parameter, this could be implemented as:
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 事件系统的方法,我强烈建议您阅读 documentation on how to work with events首先,为您的特定需求选择正确的实现/机制。基于此信息,我能够在引擎需要建立新连接时生成新的 RDS IAM 密码。
关于python-3.x - 使用 Boto3 进行 IAM 身份验证的 SQLAlchemy 可刷新凭据,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/61100925/