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

wordpress get_attached_media北京网站seo哪家公司好

wzjs 2025/8/2 18:35:39
wordpress get_attached_media,北京网站seo哪家公司好,建设网站要求和注意事项,2023年建筑项目招标信息问题 最近需要改造一个Python的Flask项目。需要在这个项目中添加AWS Secrets Manager作为配置中心，主要是数据库相关配置。 前提 得预先在Amazon RDS里面新建好数据库用户和数据库，以AWS Aurora为例子，建库和建用户语句类似如下&#xff1…

问题

最近需要改造一个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.dtcms.com/wzjs/193142.html

相关文章：

  • 做家政网站公司东莞发布最新通告
  • 成都网站建设优秀公司搜索历史记录
  • 自己做头像的网站洛阳seo网络推广
  • 厦门网站建设哪家强关键词seo报价
  • 做外贸做的很好的网站爱站网关键词挖掘工具站长工具
  • 打开网站弹出一张图片 怎么做怎样把自己的产品放到网上销售
  • 天水 网站建设招聘经典软文案例200字
  • 自己做游戏资讯网站推广app赚佣金平台有哪些
  • 怎样学做网站线下推广怎么做
  • 动漫设计本科seo标题生成器
  • 网易云音乐wordpress插件楚雄seo
  • 酒店网站搜索引擎优化方案站长工具高清吗
  • 网站改版设计费进什么科目广州网络营销推广公司
  • 做网站毕设答辩问题软件开发外包平台
  • 中国标准物质信息网网站建设如何用模板做网站
  • 有些人做网站不用钱的，对吗？2345导网址导航下载
  • 网站建设与推广中国营销网站
  • 个人类网站有哪些投放广告的渠道有哪些
  • wordpress影视站怎么推广游戏代理赚钱
  • 哪个企业的网站做的比较好成都网络推广中联无限
  • 郑州航空港建设局网站关键词代发排名首页
  • 百度搜寻网站缩略图如何添加遵义网站seo
  • 网站建设技术大全seo是什么职位缩写
  • bbc实验是哪个网站做的郑州百度快照优化排名
  • 营销型网站建设菲凡网举例说明什么是seo
  • wordpress大图片谷歌seo工具
  • 怎么做跟P站一样的网站2020站群seo系统
  • 视频素材网湖南竞价优化哪家好
  • 沧州 网站建设北京网站建设公司哪家好
  • 便宜的seo网站优化排名百度商城官网