小编Chr*_*ris的帖子

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

我使用 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)

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

sqlalchemy amazon-rds python-3.x boto3

9
推荐指数
1
解决办法
4976
查看次数

标签 统计

amazon-rds ×1

boto3 ×1

python-3.x ×1

sqlalchemy ×1