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

Python Flask中启用AWS Secrets Manager+AWS Parameter Store配置中心

news 2025/6/6 12:26:15

问题

最近需要改造一个Python的Flask项目。需要在这个项目中添加AWS Secrets Manager作为配置中心，主要是数据库相关配置。

前提

得预先在Amazon RDS里面新建好数据库用户和数据库，以AWS Aurora为例子，建库和建用户语句类似如下：

create database hf_gc default character set utf8mb4 collate utf8mb4_unicode_ci;
create user 'hf_gc'@'%' identified by '3pbP0EMfZcOusvaawfaOv4U';
grant all privileges on hf_gc.* to 'hf_gc'@'%';
flush privileges;

Secrets Manager

新建轮转密钥

开始新建Aurora密钥，如下图：
Aurora
设置密钥名称为/dev/flask-rds，如下图：
设置密钥名称
启用自动轮转密码，并设置相关策略，如下图：
轮换策略
最后审计一下，没问题就直接创建，如下图：
创建审计

Parameter Store

设置数据库主机参数，如下图：
设置数据库主机参数
设置Flask Web应用程序使用的轮转密钥参数，如下图：
设置localhost环境使用dev的数据库

Flask

FLASK_PROFILE自定义环境变量

设置自定义环境变量:FLASK_PROFILE。内容如下：

export FLASK_PROFILE="dev"

这里需要以FLASK_开头的，进行环境变量命名的。
PyCharm设置如下图：
IDE中设置环境变量
在Flask中读取环境变量，类似如下方式：

app.config.from_prefixed_env()
app.config["FLASK_PROFILE"]  # Is "dev"

安装boto3 SDK

pip install boto3
pip install botocore

secrets_manager.py

aws secrets manager客户端工具类，这里主要就是使用boto3来访问AWS 的secrets manager服务，然后取secrets manager服务中配置/dev/flask-rds参数。还使用lru_cache缓存管理注解，在使用的时候，需要进行缓存处理。内容如下：

import json
import sys
sys.path.append('./')
from typing import Dict, Anyimport boto3
from botocore.exceptions import ClientError
from functools import lru_cache
from flask import current_appclass SecretsManagerClient:def __init__(self, region_name: str = 'cn-north-1'):"""初始化客户端:param region_name: AWS区域，默认为中国区北京"""self.session = boto3.session.Session()self.client = self.session.client(service_name='secretsmanager',region_name=region_name)@lru_cache(maxsize=32)  # 缓存最多32个不同的secretdef get_secret(self, secret_name: str) -> Dict[str, Any]:"""获取Secret的值（带缓存）:param secret_name: Secret的完整路径或名称:return: 解析后的Secret字典:raises: ClientError 当获取失败时抛出"""try:current_app.logger.debug(f"Fetching secret: {secret_name}")response = self.client.get_secret_value(SecretId=secret_name)if 'SecretString' in response:secret = response['SecretString']return json.loads(secret)else:raise ValueError("Binary secrets are not supported")except ClientError as e:current_app.logger.error(f"Failed to get secret {secret_name}: {str(e)}")raiseexcept json.JSONDecodeError as e:current_app.logger.error(f"Failed to parse secret JSON: {str(e)}")raise ValueError("Invalid secret JSON format")def clear_cache(self, secret_name: str = None):"""清除缓存:param secret_name: 指定清除某个secret的缓存，None表示清除全部"""if secret_name:self.get_secret.cache_clear(secret_name)else:self.get_secret.cache_clear()current_app.logger.debug(f"Cleared cache for secret: {secret_name or 'all'}")# 创建全局客户端实例（单例模式）
client = SecretsManagerClient()

使用方式，类似如下代码：

from configs.secrets_manager import client
client.clear_cache("/dev/flask-rds")  # 清除缓存
secret = client.get_secret("/dev/flask-rds")
user = secret['username']  # 读取用户名
passwd = secret['password']

parameter_store.py

parameter store工具类内容如下：

import json
import sys
sys.path.append('./')
from typing import Dict, Anyimport boto3
from botocore.exceptions import ClientError
from functools import lru_cache
from flask import current_appclass SecretsManagerClient:def __init__(self, region_name: str = 'cn-north-1'):"""初始化客户端:param region_name: AWS区域，默认为中国区北京"""self.session = boto3.session.Session()self.client = self.session.client(service_name='secretsmanager',region_name=region_name)@lru_cache(maxsize=32)  # 缓存最多32个不同的secretdef get_secret(self, secret_name: str) -> Dict[str, Any]:"""获取Secret的值（带缓存）:param secret_name: Secret的完整路径或名称:return: 解析后的Secret字典:raises: ClientError 当获取失败时抛出"""try:current_app.logger.debug(f"Fetching secret: {secret_name}")response = self.client.get_secret_value(SecretId=secret_name)if 'SecretString' in response:secret = response['SecretString']return json.loads(secret)else:raise ValueError("Binary secrets are not supported")except ClientError as e:current_app.logger.error(f"Failed to get secret {secret_name}: {str(e)}")raiseexcept json.JSONDecodeError as e:current_app.logger.error(f"Failed to parse secret JSON: {str(e)}")raise ValueError("Invalid secret JSON format")def clear_cache(self, secret_name: str = None):"""清除缓存:param secret_name: 指定清除某个secret的缓存，None表示清除全部"""if secret_name:self.get_secret.cache_clear(secret_name)else:self.get_secret.cache_clear()current_app.logger.debug(f"Cleared cache for secret: {secret_name or 'all'}")# 创建全局客户端实例（单例模式）
client = SecretsManagerClient()

使用方式，类似如下代码：

from parameter_store import ssmClient
ssmClient.clear_cache("/local_flask/db/secret_name")  # 清除缓存
secret_name = ssmClient.get_parameter("/local_flask/db/secret_name")

数据库场景测试

在Flask Web应用程序中，获取数据库连接类，类似内容如下：

# -*- coding: utf-8 -*-import sys
sys.path.append('./')
import pymysql
from secrets_manager import client
from parameter_store import ssmClientfrom tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from flask import current_app# 数据库连接
@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1, min=4, max=10),retry=(retry_if_exception_type(pymysql.OperationalError) |retry_if_exception_type(pymysql.InternalError)))
def get_conn_sqldb():conn = Nonesecret_name_title = "/{}_flask/db/secret_name".format(current_app.config['PROFILE'])secret_name = ssmClient.get_parameter(secret_name_title)db_host_title = "/{}_flask/db/host".format(current_app.config['PROFILE'])try:secret = client.get_secret(secret_name)host = ssmClient.get_parameter(db_host_title)port = secret['port']  # 端口号user = secret['username']  # 用户名passwd = secret['password']db = 'xxx_database'  # 库名current_app.logger.debug(f"Connecting to {host}:{port}")conn = pymysql.connect(host=host, port=port, user=user,passwd=passwd, db=db, charset='utf8mb4', client_flag=pymysql.constants.CLIENT.MULTI_STATEMENTS)return connexcept pymysql.OperationalError as e:if conn:conn.close()if e.args[0] == 1045:  # Access deniedcurrent_app.logger.warning("Authentication failed, refreshing secret...")client.clear_cache(secret_name)  # 清除缓存ssmClient.clear_cache(secret_name_title)  # 清除缓存ssmClient.clear_cache(db_host_title)  # 清除缓存raiseraiseexcept Exception as e:if conn:conn.close()current_app.logger.error(f"Unexpected error: {str(e)}")raise

这里使用了tenacity库的@retry注解，进行数据库连接失败时重试策略；还使用了current_app.config['PROFILE']方式来获取Flask环境变量配置；还针对AWS RDS响应1045编码做了清除缓存处理。这里主要是先从parameters store中获取secret名称后，再拿个这个值，去secrets manager换RDS轮换密钥数据库密码，数据库端口，数据库用户名。

总结

现在这个点（2025年），Flask在Python里面的地位。貌似没有Spring生态在java中Web地位那么强势和全面。AWS这两个服务可以统一都通过boto3 SDK访问，比以前强多了，以前parameters store和secrets manager这个两个服务，客户端得安装两个SDK才能读取数据，有点费劲，这一点比以前强多了。这里直接使用Python自带的@lru_cache注解感觉挺香。重试库Tenacity也挺香的。

参考

  • Configuring from Environment Variables
  • AWS Systems Manager Parameter Store
  • get_parameter
  • How to Use SSM Parameter Store with boto3
  • Logging
  • How to Get Started with Logging in Flask
  • Tenacity
  • 使用 Python AWS SDK 获取 Secrets Manager 密钥值
  • 在不重启容器的情况下轮换数据库凭证
http://www.xdnf.cn/news/863281.html

相关文章：

  • Go语言爬虫系列教程4：使用正则表达式解析HTML内容
  • dvwa9——Weak Session IDs
  • Redis-旁路缓存策略详解
  • 常见排序算法详解与C语言实现
  • Python网页数据抓取常用的库及方法介绍
  • Python非监督学习
  • 如何轻松地将文件从 PC 传输到 iPhone?
  • 吃透 Golang 基础：数据结构之 Struct
  • 涂胶协作机器人解决方案 | Kinova Link 6 Cobot在涂胶工业的方案应用与价值
  • 四、函数调用包含单个参数之Double类型-mmword，movsd，mulsd，addsd指令，总结汇编的数据类型
  • 4.1 HarmonyOS NEXT原生AI能力集成：盘古大模型端侧部署与多模态交互实战
  • 在compose中的Canvas用kotlin显示多数据波形闪烁的问题
  • 李飞飞World Labs开源革命性Web端3D渲染器Forge！3D高斯溅射技术首次实现全平台流畅运行
  • VR博物馆推动现代数字化科技博物馆
  • 【Linux】进程 信号保存 信号处理 OS用户态/内核态
  • bug：undefined is not iterable (cannot read property Symbol(Symbol.iterator))
  • Flutter面试题
  • 【学习笔记】Circuit Tracing: Revealing Computational Graphs in Language Models
  • 【OSG学习笔记】Day 15: 路径动画与相机漫游
  • 海康网络摄像头实时取帧转Opencv数组格式(h,w,3)，已实现python、C#
  • [C]C语言日志系统宏技巧解析
  • 让视觉基础模型（VFMs）像大语言模型（LLMs）一样“会思考”​
  • 3D视觉重构工业智造：解码迁移科技如何用“硬核之眼“重塑生产节拍
  • SOLIDWORKS 2025教育版提供了丰富的自学资源
  • LabVIEW与Modbus/TCP温湿度监控系统
  • 扫地机产品--材质传感器算法开发与虚拟示波器
  • R语言基础| 下载、安装
  • Elasticsearch中的文档（Document）介绍
  • 代码随想录算法训练营第60期第五十七天打卡
  • Elasticsearch从安装到实战、kibana安装以及自定义IK分词器/集成整合SpringBoot详细的教程（二）