当前位置：首页 > news >正文

AWS Lambda Python + AWS Secrets Manager + AWS Aurora Mysql

news 2025/6/12 18:28:01

目标

在AWS Lambda Python 中根据AWS Secrets Manager 密码查询AWS Aurora Mysql数据库数据。

AWS Secrets Manager集成

前提

假设已经配置RDS（ AWS Aurora Mysql）数据库的轮转密钥在AWS Secrets Manager中。

GetSecretValue读取权限策略

给lambda执行角色配置GetSecretValue读取AWS Secrets Manager权限策略，具体权限如下：

{"Version": "2012-10-17","Statement": [{"Sid": "VisualEditor0","Effect": "Allow","Action": ["secretsmanager:GetSecretValue","secretsmanager:DescribeSecret"],"Resource": "arn:aws-cn:secretsmanager:*:2342423:secret:*"}]
}

lambda角色位置，如下图：

Requests依赖层

因为我们的lambda程序访问Secrets Manager使用到了Requests依赖库，所以这里给lambda配置相关依赖层，具体过程如下：

# 在本地开发机器上面创建一个python3.10的虚拟环境
conda create -n aws-lambda-py-3.10 python=3.10
# 激活虚拟环境
conda activate aws-lambda-py-3.10
# 基于上面的虚拟环境开始下载requests依赖库
pip install --target ./python requests==2.32.3
# 压缩python文件夹，制作lambda的requests依赖层，压缩文件，需要排除额外的文件属性
zip -r -X requests2.32.3.zip ./python
# 检查zip包
zip -sf requests2.32.3.zip

制作好requests依赖层后，将requests2.32.3.zip包上传到lambda层管理中，层创建入口如下图：

添加依赖层

主要是给lambda添加两个依赖层即可：

requests依赖库：作为http客户端依赖库；
AWS-Parameters-and-Secrets-Lambda-Extension依赖库：AWS Parameters和Secrets两个服务对lambda函数的扩展支持依赖库，他是AWS官方自己维护的依赖库。

添加完成的结果如下图：
依赖库层

AWS Aurora Mysql集成

由于Lambda函数需要查询数据库中的数据，所以，还需要集成Python相关的ORM框架进行处理。这里选择使用sqlalchemy框架。

sqlalchemy依赖层

# 下载依赖库
pip install --target ./python SQLAlchemy==2.0.41
# 制作zip包
zip -r -X SQLAlchemy2.0.41.zip ./python
# 检查zip包
zip -sf SQLAlchemy2.0.41.zip

制作依赖层没问题后，添加这个依赖层，如下图：
开始添加依赖层
创建依赖层，如下图：
依赖层创建
下面开始添加SQLAlchemy库到需要使用到lambda函数中，具体如下图：
开始添加SQLAlchemy依赖层
添加SQLAlchemy依赖层即可，如下图：
添加自定义依赖层
结果如下：
现在依赖层状态

pymysql依赖层

# 下载依赖库
pip install --target ./python PyMySQL==1.1.1
# 制作zip包
zip -r -X PyMySQL1.1.1.zip ./python
# 检查zip包
zip -sf PyMySQL1.1.1.zip

现在lambda依赖层状态如下：
pymysql依赖库

附加vpc

由于AWS Aurora Mysql只能通过vpc内网访问，所以，lambda需要附加vpc，通过内网访问数据库。

添加AWSLambdaVPCAccessExecutionRole权限策略

找到lambda函数执行角色，给这个角色添加AWSLambdaVPCAccessExecutionRole的权限策略，这个权限策略是AWS官方托管维护的。最终lambda函数执行角色，权限如下：
添加vpc执行权限策略

lambdaForVPC自定义权限策略

lambda函数需要验证vpc中的验证网络资源权限，需要自定义这个权限策略，内容如下：

{"Version": "2012-10-17","Statement": [{"Sid": "lambdaForVPC","Effect": "Allow","Action": ["ec2:DescribeSecurityGroups","ec2:DescribeSubnets","ec2:DescribeVpcs","ec2:getSecurityGroupsForVpc"],"Resource": "*"}]
}

lambda执行角色权限如下：
添加自定义权限策略

开始附加vpc

如下图：
开始附加vpc
设置vpc安全组配置，这里直接使用vpc默认安全组了，如下图：

等待一段时间完成创建即可。

Lambda Python

# encoding: UTF-8
import json
import os
import requests
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
from typing import Dict, Anyfrom sqlalchemy import Column, BigInteger, String, Text, Integer, Date, JSON
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy import selectimport loggingprint('Loading function')
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)def get_db_secret(secret_arn: str) -> Dict[str, Any]:"""从AWS Secrets Manager获取数据库凭据"""try:secrets_extension_endpoint = (f"http://localhost:2773/secretsmanager/get?secretId={secret_arn}")headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}response = requests.get(secrets_extension_endpoint, headers=headers)response.raise_for_status()  # 如果请求失败会抛出HTTPErrorsecret_data = json.loads(response.text)return json.loads(secret_data["SecretString"])except requests.exceptions.RequestException as e:print(f"请求Secrets Manager失败: {str(e)}")raiseexcept (KeyError, json.JSONDecodeError) as e:print(f"解析Secret失败: {str(e)}")raisedef create_db_engine(secret: Dict[str, Any]) -> sqlalchemy.engine.Engine:"""创建SQLAlchemy数据库引擎"""try:db_url = (f"mysql+pymysql://{secret['username']}:{secret['password']}@"f"{secret['host']}:{secret['port']}/hf_gc""?charset=utf8mb4&binary_prefix=true")return create_engine(db_url,pool_recycle=3600,pool_pre_ping=True  # 添加连接健康检查)except KeyError as e:print(f"缺少必要的数据库连接参数: {str(e)}")raiseexcept sqlalchemy.exc.SQLAlchemyError as e:print(f"创建数据库引擎失败: {str(e)}")raiseclass Base(DeclarativeBase):passclass User(Base):__tablename__ = "user_account"id: Mapped[int] = mapped_column(primary_key=True)name: Mapped[str] = mapped_column(String(30))fullname: Mapped[Optional[str]]def lambda_handler(event, context):try:# Replace with the name or ARN of your secretsecret_arn = "arn:aws-cn:secretsmanager:cn-north-1:23546578098765:secret:/dev/flask-rds-zgprXn"db_secret = get_db_secret(secret_arn)# 2. 创建数据库连接engine = create_db_engine(db_secret)session = Session(engine)stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))# print("原始 SQL:", str(stmt.compile(engine, compile_kwargs={"literal_binds": True})))for user in session.scalars(stmt):print(user)return {'statusCode': 200,'body': json.dumps({'message': 'push to bc OK'})}except Exception as e:print(f"Error: {str(e)}")return {'statusCode': 500,'body': json.dumps({'message': 'Error retrieving secret','error': str(e)})}