我使用 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)
Run Code Online (Sandbox Code Playgroud)
问题是提供的令牌仅在 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
)
Run Code Online (Sandbox Code Playgroud)
是否有内置函数,或者有更优雅的方法来解决这个问题?