AWS Lambda用PyMySQL可连RDS，SQLAlchemy连接失败求助

问题根源在于IAM认证生成的database_token包含特殊字符（如/、+、=等），直接拼接进SQLAlchemy的连接URL时，会触发自动URL编码解析，导致密码部分被截断或篡改，最终出现using password: NO的错误——SQLAlchemy实际未接收到正确的密码。而PyMySQL直接传入token字符串不会触发编码逻辑，因此能正常连接。

解决方法 #

方法1：使用SQLAlchemy官方URL构造器（推荐） #

用sqlalchemy.engine.URL.create替代手动拼接字符串，它会自动处理特殊字符的编码，避免解析错误：

from sqlalchemy import create_engine
from sqlalchemy.engine import URL
import os
import boto3

rds_client = boto3.client('rds')
database_token = rds_client.generate_db_auth_token(
    DBHostname=os.environ["DB_HOST"],
    Port=3306,
    DBUsername=os.environ["DB_USER"],
    Region=os.environ["AWS_REGION"],
)

# 通过URL构造器生成合规的连接地址
db_url = URL.create(
    drivername="mysql+pymysql",
    username=os.environ["DB_USER"],
    password=database_token,
    host=os.environ["DB_HOST"],
    port=3306,
    database=os.environ["DB_NAME"]
)

engine = create_engine(db_url, connect_args={'ssl': {"any_non_empty_dict": True}}, echo=True)
database_connection = engine.connect()

方法2：手动URL编码token #

如果坚持手动拼接URL，需用urllib.parse.quote_plus对token进行URL编码，确保特殊字符被正确转义：

from sqlalchemy import create_engine
import os
import boto3
from urllib.parse import quote_plus

rds_client = boto3.client('rds')
database_token = rds_client.generate_db_auth_token(
    DBHostname=os.environ["DB_HOST"],
    Port=3306,
    DBUsername=os.environ["DB_USER"],
    Region=os.environ["AWS_REGION"],
)

# 编码token中的特殊字符
encoded_token = quote_plus(database_token)
DB_URL = f"mysql+pymysql://{os.environ['DB_USER']}:{encoded_token}@{os.environ['DB_HOST']}:3306/{os.environ['DB_NAME']}"

engine = create_engine(DB_URL, connect_args={'ssl': {"any_non_empty_dict": True}}, echo=True)
database_connection = engine.connect()

额外验证点 #

• 确认Lambda的IAM角色已配置rds-db:connect权限，且关联了目标RDS资源（你用PyMySQL能正常连接，此步骤大概率无问题）
• 确保目标RDS实例已启用IAM认证，且数据库用户绑定了对应的IAM角色

内容的提问来源于stack exchange，提问作者Ben Slater