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

广东品牌网站建设服务网站推广方案

wzjs 2025/7/19 2:11:13

广东品牌网站建设,服务网站推广方案,北京最新消息今天上午,河南做网站河南网站建设目标在AWS Lambda Python 中根据AWS Secrets Manager 密码查询AWS Aurora Mysql数据库数据。 AWS Secrets Manager集成前提假设已经配置RDS（ AWS Aurora Mysql）数据库的轮转密钥在AWS Secrets Manager中。 GetSecretValue读取权限策略给lambda…

目标

在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)})}