如何在AWS Lambda环境中实现Python SqlModel与AWS RDS的连接?
如何在AWS Lambda环境中实现Python SqlModel与AWS RDS的连接?
我之前也碰到过一模一样的问题!本地用SqlModel连RDS顺得不行,一迁到Lambda就各种连不上,折腾了好几天才摸清门道,给你分享几个关键的排查点和解决方案:
一、先搞定Lambda与RDS的网络连通性(最容易踩坑的环节)
- 如果你把Lambda部署在VPC内,必须确保Lambda所在的子网能访问RDS的安全组:
- 给RDS的安全组添加入站规则,允许Lambda所在子网的IP段(或者直接关联Lambda的安全组)访问3306端口(MySQL默认端口)。
- 如果Lambda没放在VPC里,那RDS得开启公网访问,同时安全组暂时允许
0.0.0.0/0入站(生产环境不建议这么做,优先用VPC部署)。
- 可以在Lambda里加一段测试代码,快速验证网络是否通畅:
import socket def test_rds_connection(): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) # 替换成你的RDS端点和端口 result = sock.connect_ex(('your-rds-endpoint.rds.amazonaws.com', 3306)) if result == 0: print("✅ RDS端口可访问") else: print("❌ RDS端口无法访问") sock.close() except Exception as e: print(f"❌ 连接测试出错: {str(e)}")
- 另外,Lambda的执行角色必须拥有
ec2:CreateNetworkInterface、ec2:DescribeNetworkInterfaces、ec2:DeleteNetworkInterface这三个权限,不然它连VPC内的网络资源都创建不了。
二、解决部署包的依赖适配问题
SqlModel依赖SQLAlchemy、pymysql这些库,本地Windows/Mac安装的依赖在Lambda的Linux环境下可能跑不起来,一定要用适配Lambda运行环境的依赖包:
- 推荐用Docker打包:拉取AWS官方的Python构建镜像(比如
public.ecr.aws/sam/build-python3.10),在容器里安装依赖,再把site-packages里的内容打包出来。 - 或者用Lambda层:把
sqlmodel、sqlalchemy、pymysql打包成一个层,附加到你的Lambda函数上,这样多个函数可以复用依赖,也能减小部署包体积。 - 注意:不要直接把本地的
.venv文件夹打包进去,只需要site-packages里的核心依赖文件。
三、调整SqlModel的连接配置(关键细节)
本地的连接字符串在Lambda里要做两个关键调整:
- 替换为RDS的端点地址
- 加上SSL证书参数(AWS RDS强制要求SSL连接)
给你一个可直接复用的代码示例:
import os from sqlmodel import SQLModel, create_engine, Session # 把引擎定义在函数外部,复用连接池(Lambda是无状态的,避免每次调用都新建连接) engine = None def init_db_engine(): global engine if engine is None: # 从Lambda环境变量读取数据库配置(推荐用Secrets Manager存敏感信息,下文会说) db_user = os.environ["DB_USER"] db_password = os.environ["DB_PASSWORD"] db_host = os.environ["DB_HOST"] db_name = os.environ["DB_NAME"] # 关键:加上charset和ssl_ca参数,RDS MySQL必须用SSL连接 # 记得把rds-ca-2019-root.pem下载到部署包里,路径要正确 connection_string = ( f"mysql+pymysql://{db_user}:{db_password}@{db_host}:3306/{db_name}" "?charset=utf8mb4&ssl_ca=rds-ca-2019-root.pem" ) # pool_recycle设置为300秒,避免RDS自动断开闲置连接 engine = create_engine(connection_string, echo=False, pool_recycle=300) # 第一次调用时自动创建表(如果需要初始化表结构) SQLModel.metadata.create_all(engine) def lambda_handler(event, context): try: init_db_engine() with Session(engine) as session: # 这里写你的数据库操作逻辑,比如查询 # example_user = session.exec(select(User).where(User.id == 1)).first() return {"statusCode": 200, "body": "✅ 数据库连接成功!"} except Exception as e: return {"statusCode": 500, "body": f"❌ 连接失败: {str(e)}"}
四、用Secrets Manager管理敏感信息(更安全)
不要把数据库密码明文放在Lambda环境变量里,推荐用AWS Secrets Manager存储:
- 先在Secrets Manager里创建一个RDS数据库凭证的密钥,包含用户名、密码、端点、数据库名。
- 给Lambda的执行角色添加
secretsmanager:GetSecretValue权限。 - 然后修改代码读取密钥:
import boto3 import json def get_db_credentials(): secrets_client = boto3.client('secretsmanager') secret_response = secrets_client.get_secret_value(SecretId='your-secret-name') secret_data = json.loads(secret_response['SecretString']) return ( secret_data['username'], secret_data['password'], secret_data['host'], secret_data['dbname'] )
五、利用CloudWatch日志快速排查问题
如果还是连不上,一定要去Lambda的CloudWatch日志里看具体错误:
- 要是看到
OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on 'xxx' (timed out)"):网络不通,回到第一步检查VPC和安全组。 - 要是看到
ModuleNotFoundError: No module named 'sqlmodel':依赖没打包对,检查部署包或Lambda层。 - 要是看到
SSLError: SSL connection error: certificate verify failed:SSL证书路径不对或者没打包,重新下载rds-ca-2019-root.pem放到部署包里。
备注:内容来源于stack exchange,提问作者aadhil96




