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

问题

最近需要改造一个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密钥，如下图：

设置密钥名称为/dev/flask-rds，如下图：

启用自动轮转密码，并设置相关策略，如下图：

最后审计一下，没问题就直接创建，如下图：

Parameter Store

设置数据库主机参数，如下图：

设置Flask Web应用程序使用的轮转密钥参数，如下图：

Flask

FLASK_PROFILE自定义环境变量

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

export FLASK_PROFILE="dev"

这里需要以FLASK_开头的，进行环境变量命名的。
PyCharm设置如下图：

在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 密钥值
• 在不重启容器的情况下轮换数据库凭证