如何使用token通过SQLAlchemy连接AWS RDS MySQL并解决参数报错问题

问题原因 #

你遇到的报错是两处参数混淆导致的：

1. cparams['token']是部分数据库驱动支持的专属参数，你当前使用的mysql+pymysql驱动的连接初始化参数中不存在token字段，因此会报参数未预期的错误。
2. attrs_before是SQL Server专属的连接配置参数，完全不适用于MySQL场景，无需尝试该方案。

正确实现方案 #

AWS RDS MySQL的IAM身份验证生成的token本质是临时密码，你只需要将生成的token作为密码传入pymysql的连接参数，同时配置好强制SSL连接即可正常连接。

修正后的可运行代码如下：

import boto3
from sqlalchemy import create_engine, event
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import sessionmaker

# 初始化AWS客户端
session = boto3.session.Session(profile_name='xxx')
rds = session.client('rds')

# RDS连接配置，替换为你的实际值
RDS_HOST = 'xxxx'
RDS_PORT = 3306
RDS_USER = 'xxx'
RDS_DB_NAME = '你的数据库名'
# 替换为你本地存放的AWS RDS根证书路径
RDS_SSL_CA_PATH = '/path/to/rds-ca-root.pem'

def get_authentication_token():
    return rds.generate_db_auth_token(
        DBHostname=RDS_HOST, 
        Port=RDS_PORT, 
        DBUsername=RDS_USER
    )

# 构造数据库连接URL
engine_url = URL.create(
    drivername='mysql+pymysql',
    host=RDS_HOST,
    port=RDS_PORT,
    username=RDS_USER,
    database=RDS_DB_NAME
)
engine = create_engine(engine_url)

@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    # 将token作为密码传入pymysql的passwd参数
    cparams['passwd'] = get_authentication_token()
    # 配置SSL连接，IAM认证强制要求SSL
    cparams['ssl'] = {'ca': RDS_SSL_CA_PATH}

Session = sessionmaker(bind=engine)

with Session() as session:
    result = session.execute('select now()').first()
    print(result)

注意事项 #

• 提前为你的RDS MySQL用户开启IAM身份验证权限，对应SQL语句为：ALTER USER 'xxx'@'%' IDENTIFIED WITH AWSAuthenticationPlugin AS 'RDS';
• 确保你的AWS身份凭证拥有rds:GenerateDbAuthToken权限
• 生成token时传入的Host、Port、用户名必须和实际连接时的参数完全一致，否则会认证失败

内容的提问来源于stack exchange，提问作者This-is-patriiick