配置管理在软件开发中至关重要：
开发、测试、生产环境使用不同配置
敏感信息（API密钥、数据库密码）安全存储
无需修改代码即可改变应用行为
| 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
app.config直接设置 |
简单直观 | 硬编码，不安全 | 快速原型，临时测试 |
| 配置类 | 结构清晰，支持继承 | 需要Python代码 | 大多数Flask项目 |
| 配置文件（.py, .cfg） | 配置与代码分离 | 可能有安全风险 | 不包含敏感信息的配置 |
| 环境变量 | 安全，便于部署 | 管理复杂 | 敏感信息，生产环境 |
| 配置管理工具 | 集中管理，版本控制 | 学习成本 | 大型微服务架构 |
# 最基本的配置方式
from flask import Flask
app = Flask(__name__)
# 直接设置配置项
app.config['SECRET_KEY'] = 'hardcoded-secret-key'
app.config['DEBUG'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SESSION_COOKIE_SECURE'] = False # 开发环境可以设置为False
app.config['PERMANENT_SESSION_LIFETIME'] = 3600 # 1小时
# 批量更新配置
app.config.update(
TESTING=True,
PROPAGATE_EXCEPTIONS=True
)
# 从字典加载配置
config_dict = {
'MAX_CONTENT_LENGTH': 16 * 1024 * 1024, # 16MB文件上传限制
'UPLOAD_FOLDER': '/path/to/uploads',
'ALLOWED_EXTENSIONS': {'txt', 'pdf', 'png', 'jpg'}
}
app.config.from_mapping(config_dict)密钥、密码等敏感信息永远不要直接写在代码中，应该使用环境变量或安全的配置管理系统。
# config.py - 配置类定义
import os
from datetime import timedelta
class Config:
"""基础配置类"""
# 应用元数据
APP_NAME = 'My Flask App'
APP_VERSION = '1.0.0'
# 安全配置
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
# 会话配置
SESSION_COOKIE_NAME = 'flask_session'
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SECURE = False # 生产环境应为True
PERMANENT_SESSION_LIFETIME = timedelta(days=7)
# 文件上传
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER = 'uploads'
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}
# 数据库配置（基础）
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 性能配置
JSON_SORT_KEYS = False # 不排序JSON键，提高性能
JSONIFY_PRETTYPRINT_REGULAR = False # 生产环境关闭美化
@staticmethod
def init_app(app):
"""初始化应用"""
pass
class DevelopmentConfig(Config):
"""开发环境配置"""
DEBUG = True
TESTING = False
# 数据库
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///dev.db'
# 详细日志
LOG_LEVEL = 'DEBUG'
# 开发工具
EXPLAIN_TEMPLATE_LOADING = False
# 禁用缓存
SEND_FILE_MAX_AGE_DEFAULT = 0
# 跨域支持（开发时可能用到）
CORS_ORIGINS = ['http://localhost:3000', 'http://127.0.0.1:3000']
class TestingConfig(Config):
"""测试环境配置"""
DEBUG = False
TESTING = True
# 测试数据库
SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
'sqlite:///:memory:' # 内存数据库
# 测试特定配置
WTF_CSRF_ENABLED = False # 测试时禁用CSRF
PRESERVE_CONTEXT_ON_EXCEPTION = False
# 日志级别
LOG_LEVEL = 'INFO'
# 邮件测试
MAIL_SUPPRESS_SEND = True # 不实际发送邮件
class ProductionConfig(Config):
"""生产环境配置"""
DEBUG = False
TESTING = False
# 生产数据库
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
# 安全增强
SESSION_COOKIE_SECURE = True
REMEMBER_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
# 生产环境必须设置SECRET_KEY
SECRET_KEY = os.environ['SECRET_KEY'] # 强制要求设置
# 性能优化
JSONIFY_PRETTYPRINT_REGULAR = False
SEND_FILE_MAX_AGE_DEFAULT = 31536000 # 1年缓存
# 日志级别
LOG_LEVEL = 'WARNING'
# 服务器配置
PREFERRED_URL_SCHEME = 'https'
SERVER_NAME = os.environ.get('SERVER_NAME')
# 限流配置
RATELIMIT_ENABLED = True
RATELIMIT_DEFAULT = "200 per day;50 per hour"
@classmethod
def init_app(cls, app):
"""生产环境初始化"""
Config.init_app(app)
# 生产环境特定的初始化
cls.setup_logging(app)
cls.setup_monitoring(app)
cls.setup_error_handling(app)
@staticmethod
def setup_logging(app):
"""配置生产环境日志"""
import logging
from logging.handlers import RotatingFileHandler
# 创建日志目录
log_dir = 'logs'
os.makedirs(log_dir, exist_ok=True)
# 文件处理器
file_handler = RotatingFileHandler(
os.path.join(log_dir, 'app.log'),
maxBytes=10*1024*1024, # 10MB
backupCount=10
)
file_handler.setLevel(logging.WARNING)
# 格式化器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s '
'[in %(pathname)s:%(lineno)d]'
)
file_handler.setFormatter(formatter)
app.logger.addHandler(file_handler)
@staticmethod
def setup_monitoring(app):
"""配置监控"""
# 这里可以集成Sentry、New Relic等监控工具
pass
@staticmethod
def setup_error_handling(app):
"""配置错误处理"""
# 生产环境禁用调试信息
app.config['PROPAGATE_EXCEPTIONS'] = False
app.config['TRAP_HTTP_EXCEPTIONS'] = True
# 配置映射字典
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
# 在应用中加载配置
def create_app(config_name=None):
"""应用工厂函数"""
app = Flask(__name__)
# 确定配置
if config_name is None:
config_name = os.environ.get('FLASK_ENV', 'default')
# 加载配置
app.config.from_object(config[config_name])
# 初始化应用
config[config_name].init_app(app)
return app# 从Python文件加载
app.config.from_pyfile('config.py')
# 从JSON文件加载
app.config.from_json('config.json')
# 从环境变量指定的文件加载
app.config.from_envvar('FLASK_CONFIG_FILE')
# 示例配置文件：config/production.py
"""
生产环境配置文件
注意：此文件不应包含真正的密钥，密钥应从环境变量读取
"""
import os
# 基础配置
DEBUG = False
TESTING = False
SECRET_KEY = os.environ['SECRET_KEY']
# 数据库配置
SQLALCHEMY_DATABASE_URI = os.environ['DATABASE_URL']
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10,
'pool_recycle': 300,
'pool_pre_ping': True,
'max_overflow': 20
}
# Redis配置
REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
CACHE_TYPE = 'redis'
CACHE_REDIS_URL = REDIS_URL
# 邮件配置
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = True
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER')
# 文件上传
MAX_CONTENT_LENGTH = 50 * 1024 * 1024 # 50MB
UPLOAD_FOLDER = '/var/www/uploads'
ALLOWED_EXTENSIONS = {'jpg', 'jpeg', 'png', 'gif', 'pdf', 'doc', 'docx'}
# 安全配置
SESSION_COOKIE_SECURE = True
REMEMBER_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
# 性能配置
JSON_SORT_KEYS = False
JSONIFY_PRETTYPRINT_REGULAR = False
SEND_FILE_MAX_AGE_DEFAULT = 31536000 # 1年
# 第三方服务API密钥
GOOGLE_MAPS_API_KEY = os.environ.get('GOOGLE_MAPS_API_KEY')
STRIPE_SECRET_KEY = os.environ.get('STRIPE_SECRET_KEY')
STRIPE_PUBLIC_KEY = os.environ.get('STRIPE_PUBLIC_KEY')
# 应用特定配置
ITEMS_PER_PAGE = 20
PASSWORD_RESET_TIMEOUT = 3600 # 1小时
MAX_LOGIN_ATTEMPTS = 5# .env 文件示例（开发环境）
# 注意：.env文件不应提交到版本控制系统
# 基础配置
FLASK_APP=app.py
FLASK_ENV=development
FLASK_DEBUG=1
# 安全配置
SECRET_KEY=dev-secret-key-change-in-production
SECURITY_PASSWORD_SALT=dev-salt-change-in-production
# 数据库配置
DATABASE_URL=postgresql://user:password@localhost:5432/dev_db
REDIS_URL=redis://localhost:6379/0
# 邮件配置（开发环境可以使用假邮箱）
MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USE_TLS=1
MAIL_USERNAME=your-email@gmail.com
MAIL_PASSWORD=your-app-password
MAIL_DEFAULT_SENDER=noreply@example.com
# 第三方服务（开发环境可以使用测试密钥）
GOOGLE_MAPS_API_KEY=AIzaSyDevTestKeyChangeInProduction
STRIPE_SECRET_KEY=sk_test_51...dev
STRIPE_PUBLIC_KEY=pk_test_51...dev
# 应用特定配置
ITEMS_PER_PAGE=10
DEBUG_TOOLBAR_ENABLED=1# 使用python-dotenv加载环境变量
from dotenv import load_dotenv
import os
# 加载.env文件
load_dotenv()
# 或者在Flask应用中使用
from flask import Flask
from dotenv import load_dotenv
load_dotenv() # 在创建app之前加载
app = Flask(__name__)
# 从环境变量读取配置
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL')
# 设置默认值
app.config['DEBUG'] = os.environ.get('FLASK_DEBUG', '0').lower() in ['1', 'true', 'yes']
app.config['TESTING'] = os.environ.get('FLASK_TESTING', '0').lower() in ['1', 'true', 'yes']生产环境必须正确配置以下安全选项，以防止常见的安全漏洞。
# 安全配置完整示例
class SecurityConfig:
"""安全配置"""
# === 会话安全 ===
# 加密会话数据的密钥
SECRET_KEY = os.environ['SECRET_KEY']
# 会话cookie名称
SESSION_COOKIE_NAME = 'session'
# 只允许HTTP访问cookie（防止XSS）
SESSION_COOKIE_HTTPONLY = True
# 只允许HTTPS传输cookie（生产环境必须为True）
SESSION_COOKIE_SECURE = True
# Cookie SameSite策略
# 'Strict' - 完全禁止跨站请求携带Cookie
# 'Lax' - 允许部分安全的跨站请求（推荐）
# 'None' - 允许所有跨站请求（需要Secure=True）
SESSION_COOKIE_SAMESITE = 'Lax'
# 会话有效期（秒）
PERMANENT_SESSION_LIFETIME = 3600 # 1小时
# === 记住我功能 ===
REMEMBER_COOKIE_NAME = 'remember_token'
REMEMBER_COOKIE_DURATION = timedelta(days=30)
REMEMBER_COOKIE_SECURE = True
REMEMBER_COOKIE_HTTPONLY = True
REMEMBER_COOKIE_SAMESITE = 'Lax'
# === CSRF保护 ===
# WTF扩展的CSRF配置
WTF_CSRF_ENABLED = True
WTF_CSRF_SECRET_KEY = os.environ.get('CSRF_SECRET_KEY') or SECRET_KEY
WTF_CSRF_TIME_LIMIT = 3600 # 1小时
# === 密码安全 ===
# Flask-Security-Too配置
SECURITY_PASSWORD_SALT = os.environ['SECURITY_PASSWORD_SALT']
SECURITY_PASSWORD_HASH = 'bcrypt'
SECURITY_PASSWORD_SINGLE_HASH = False
SECURITY_PASSWORD_COMPLEXITY_CHECKER = 'zxcvbn'
SECURITY_PASSWORD_MIN_LENGTH = 8
# 密码策略
SECURITY_PASSWORD_REQUIREMENTS = {
'length': 8,
'uppercase': 1,
'lowercase': 1,
'digits': 1,
'symbols': 0, # 不强制要求特殊字符
}
# 登录失败限制
SECURITY_LOGIN_WITHOUT_CONFIRMATION = False
SECURITY_TRACKABLE = True
SECURITY_MAX_LOGIN_ATTEMPTS = 5
SECURITY_LOGIN_ATTEMPTS_WINDOW = 300 # 5分钟
# === 其他安全头 ===
# 防止点击劫持
@property
def CSP_HEADERS(self):
return {
'Content-Security-Policy':
"default-src 'self'; "
"script-src 'self' https://cdn.jsdelivr.net; "
"style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
"img-src 'self' data: https:; "
"font-src 'self' https://cdn.jsdelivr.net;",
'X-Frame-Options': 'DENY',
'X-Content-Type-Options': 'nosniff',
'X-XSS-Protection': '1; mode=block',
'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
'Referrer-Policy': 'strict-origin-when-cross-origin',
}# 数据库配置完整示例
class DatabaseConfig:
"""数据库配置"""
# === SQLAlchemy配置 ===
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
# 是否追踪对象修改（生产环境应为False）
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 是否在连接池中预执行ping
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10, # 连接池大小
'pool_recycle': 3600, # 连接回收时间（秒）
'pool_pre_ping': True, # 连接前先ping
'max_overflow': 20, # 最大溢出连接数
'pool_timeout': 30, # 获取连接超时时间
'echo': False, # 是否打印SQL语句（开发环境可设为True）
}
# === 连接池配置 ===
@property
def POSTGRES_POOL_CONFIG(self):
"""PostgreSQL连接池配置"""
return {
'min_size': 1,
'max_size': 20,
'max_queries': 50000,
'max_inactive_connection_lifetime': 300.0,
}
# === Redis配置 ===
REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
# Redis作为缓存
CACHE_TYPE = 'redis'
CACHE_REDIS_URL = REDIS_URL
CACHE_DEFAULT_TIMEOUT = 300 # 5分钟
# Redis作为会话存储
SESSION_TYPE = 'redis'
SESSION_REDIS = redis.from_url(REDIS_URL)
SESSION_PERMANENT = True
SESSION_USE_SIGNER = True
# === 数据库迁移配置 ===
MIGRATIONS_DIR = 'migrations'
MIGRATIONS_AUTO_GENERATE = True
# === 备份配置 ===
@property
def BACKUP_CONFIG(self):
"""数据库备份配置"""
return {
'enabled': True,
'schedule': '0 2 * * *', # 每天凌晨2点
'retention_days': 30,
'storage_path': '/var/backups/database',
'notification_email': os.environ.get('BACKUP_NOTIFICATION_EMAIL'),
}
# === 监控配置 ===
@property
def MONITORING_CONFIG(self):
"""数据库监控配置"""
return {
'slow_query_threshold': 1.0, # 慢查询阈值（秒）
'max_connections_warning': 80, # 最大连接数警告阈值
'enable_query_log': False, # 是否记录查询日志
}# 邮件配置完整示例
class MailConfig:
"""邮件配置"""
# === 基础配置 ===
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() == 'true'
MAIL_USE_SSL = os.environ.get('MAIL_USE_SSL', 'false').lower() == 'true'
# 认证信息
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER', MAIL_USERNAME)
# === 高级配置 ===
MAIL_MAX_EMAILS = 100 # 单个连接最大发送邮件数
MAIL_ASCII_ATTACHMENTS = False # 是否使用ASCII编码附件
MAIL_SUPPRESS_SEND = False # 是否实际发送邮件（测试时设为True）
# === 连接池配置 ===
MAIL_POOL_SIZE = 5 # 连接池大小
MAIL_POOL_RECYCLE = 300 # 连接回收时间（秒）
MAIL_POOL_TIMEOUT = 30 # 获取连接超时时间
# === 邮件模板配置 ===
@property
def MAIL_TEMPLATES(self):
"""邮件模板配置"""
return {
'welcome': {
'subject': '欢迎加入{app_name}',
'template': 'emails/welcome.html',
'priority': 'normal',
},
'password_reset': {
'subject': '重置您的密码',
'template': 'emails/password_reset.html',
'priority': 'high',
'expires': 3600, # 1小时有效期
},
'notification': {
'subject': '新通知',
'template': 'emails/notification.html',
'priority': 'normal',
},
}
# === 邮件队列配置 ===
@property
def MAIL_QUEUE_CONFIG(self):
"""邮件队列配置（使用Celery或RQ）"""
return {
'enabled': True,
'queue_name': 'mail',
'retry_attempts': 3,
'retry_delay': 60, # 重试延迟（秒）
'max_retry_delay': 3600, # 最大重试延迟
}
# === 邮件监控 ===
@property
def MAIL_MONITORING(self):
"""邮件发送监控"""
return {
'track_opens': True,
'track_clicks': True,
'track_bounces': True,
'webhook_url': os.environ.get('MAIL_WEBHOOK_URL'),
}
# === 测试配置 ===
@classmethod
def get_test_config(cls):
"""获取测试配置"""
config = cls()
config.MAIL_SUPPRESS_SEND = True
config.MAIL_BACKEND = 'flask_mail.backends.testing'
return config# 常用Flask扩展配置
class ExtensionsConfig:
"""Flask扩展配置"""
# === Flask-Login ===
LOGIN_VIEW = 'auth.login'
LOGIN_MESSAGE = '请先登录以访问此页面'
LOGIN_MESSAGE_CATEGORY = 'info'
REFRESH_VIEW = 'auth.refresh'
REFRESH_MESSAGE = '会话已过期，请重新登录'
REFRESH_MESSAGE_CATEGORY = 'warning'
USE_SESSION_FOR_NEXT = True
REMEMBER_COOKIE_DURATION = timedelta(days=30)
REMEMBER_COOKIE_REFRESH_EACH_REQUEST = True
# === Flask-WTF ===
WTF_CSRF_ENABLED = True
WTF_CSRF_SECRET_KEY = os.environ.get('CSRF_SECRET_KEY')
WTF_CSRF_TIME_LIMIT = 3600
WTF_CSRF_SSL_STRICT = True # 生产环境应为True
WTF_I18N_ENABLED = True
# === Flask-CORS ===
CORS_ORIGINS = os.environ.get('CORS_ORIGINS', '*').split(',')
CORS_SUPPORTS_CREDENTIALS = True
CORS_ALLOW_HEADERS = ['Content-Type', 'Authorization', 'X-Requested-With']
CORS_EXPOSE_HEADERS = ['Content-Disposition']
CORS_MAX_AGE = 86400 # 24小时
# === Flask-Cache ===
CACHE_TYPE = 'redis'
CACHE_REDIS_URL = os.environ.get('REDIS_URL')
CACHE_DEFAULT_TIMEOUT = 300
CACHE_KEY_PREFIX = 'flask_cache:'
CACHE_THRESHOLD = 500 # 最大缓存条目数
# === Flask-Mail ===
MAIL_SERVER = os.environ.get('MAIL_SERVER')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = True
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER')
# === Flask-SocketIO ===
SOCKETIO_ASYNC_MODE = 'eventlet'
SOCKETIO_CORS_ALLOWED_ORIGINS = os.environ.get('SOCKETIO_ORIGINS', '*')
SOCKETIO_LOGGER = True
SOCKETIO_ENGINEIO_LOGGER = True
SOCKETIO_MESSAGE_QUEUE = os.environ.get('REDIS_URL')
# === Flask-RESTful ===
RESTFUL_JSON = {
'ensure_ascii': False,
'indent': None,
'separators': (',', ':'),
}
RESTFUL_ERROR_404_HELP = False # 禁用404时的默认帮助信息
# === Flask-Limiter ===
RATELIMIT_ENABLED = True
RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL')
RATELIMIT_STRATEGY = 'fixed-window'
RATELIMIT_DEFAULT = ["200 per day", "50 per hour"]
RATELIMIT_HEADERS_ENABLED = True
# === Flask-Migrate ===
MIGRATE_DIRECTORY = 'migrations'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# === Flask-Babel ===
BABEL_DEFAULT_LOCALE = 'zh_CN'
BABEL_DEFAULT_TIMEZONE = 'Asia/Shanghai'
BABEL_TRANSLATION_DIRECTORIES = 'translations'
# === Flask-Admin ===
FLASK_ADMIN_SWATCH = 'flatly'
FLASK_ADMIN_TEMPLATE_MODE = 'bootstrap3'
# === 第三方服务配置 ===
@property
def THIRD_PARTY_CONFIG(self):
"""第三方服务配置"""
return {
'stripe': {
'secret_key': os.environ.get('STRIPE_SECRET_KEY'),
'public_key': os.environ.get('STRIPE_PUBLIC_KEY'),
'webhook_secret': os.environ.get('STRIPE_WEBHOOK_SECRET'),
},
'google': {
'maps_api_key': os.environ.get('GOOGLE_MAPS_API_KEY'),
'recaptcha_secret': os.environ.get('GOOGLE_RECAPTCHA_SECRET'),
'recaptcha_site_key': os.environ.get('GOOGLE_RECAPTCHA_SITE_KEY'),
},
'aws': {
'access_key_id': os.environ.get('AWS_ACCESS_KEY_ID'),
'secret_access_key': os.environ.get('AWS_SECRET_ACCESS_KEY'),
'region': os.environ.get('AWS_REGION', 'us-east-1'),
's3_bucket': os.environ.get('AWS_S3_BUCKET'),
},
'github': {
'client_id': os.environ.get('GITHUB_CLIENT_ID'),
'client_secret': os.environ.get('GITHUB_CLIENT_SECRET'),
},
'sentry': {
'dsn': os.environ.get('SENTRY_DSN'),
'environment': os.environ.get('SENTRY_ENVIRONMENT', 'production'),
},
}# 推荐的项目结构
myapp/
├── config/
│ ├── __init__.py # 配置工厂
│ ├── base.py # 基础配置
│ ├── development.py # 开发环境配置
│ ├── testing.py # 测试环境配置
│ ├── production.py # 生产环境配置
│ └── staging.py # 预发布环境配置
├── instance/
│ └── config.py # 实例特定配置（不纳入版本控制）
├── app/
│ ├── __init__.py # 应用工厂
│ ├── models.py
│ ├── views.py
│ └── extensions.py
├── migrations/
├── tests/
├── requirements.txt
├── requirements-dev.txt
├── requirements-prod.txt
├── .env.example # 环境变量示例
├── .env # 本地环境变量（不纳入版本控制）
├── .gitignore
└── README.md使用工厂模式可以创建多个应用实例，便于测试和不同环境部署。
# config/__init__.py - 配置工厂
import os
from typing import Dict, Type, Optional
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
class Config:
"""配置基类"""
# 基础配置
APP_NAME = os.environ.get('APP_NAME', 'MyApp')
VERSION = os.environ.get('APP_VERSION', '1.0.0')
# 从环境变量读取，提供默认值
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key')
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///app.db')
@classmethod
def init_app(cls, app):
"""初始化应用"""
pass
class DevelopmentConfig(Config):
"""开发配置"""
DEBUG = True
SQLALCHEMY_ECHO = True
class TestingConfig(Config):
"""测试配置"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
class ProductionConfig(Config):
"""生产配置"""
DEBUG = False
# 生产环境必须设置环境变量
SECRET_KEY = os.environ['SECRET_KEY']
SQLALCHEMY_DATABASE_URI = os.environ['DATABASE_URL']
class ConfigFactory:
"""配置工厂"""
_configs: Dict[str, Type[Config]] = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
}
@classmethod
def get_config(cls, name: Optional[str] = None) -> Type[Config]:
"""获取配置类"""
if name is None:
name = os.environ.get('FLASK_ENV', 'development')
config_class = cls._configs.get(name.lower())
if not config_class:
raise ValueError(f"未知的配置环境: {name}")
return config_class
@classmethod
def register_config(cls, name: str, config_class: Type[Config]):
"""注册新的配置类"""
cls._configs[name.lower()] = config_class
@classmethod
def create_app_config(cls, app, config_name: Optional[str] = None):
"""创建并应用配置到应用"""
config_class = cls.get_config(config_name)
# 应用配置
app.config.from_object(config_class)
# 初始化应用
config_class.init_app(app)
return config_class
# 使用示例
from flask import Flask
def create_app(config_name=None):
app = Flask(__name__)
# 从工厂获取配置
ConfigFactory.create_app_config(app, config_name)
# 注册蓝图、扩展等
# ...
return app以下信息永远不要提交到版本控制系统：密码、API密钥、私钥、数据库连接字符串等。
# 安全的敏感信息管理
import os
from cryptography.fernet import Fernet
from base64 import b64encode, b64decode
class SecretManager:
"""安全密钥管理器"""
def __init__(self, key_file='.secrets.key'):
self.key_file = key_file
self._load_or_generate_key()
def _load_or_generate_key(self):
"""加载或生成加密密钥"""
if os.path.exists(self.key_file):
with open(self.key_file, 'rb') as f:
self.key = f.read()
else:
self.key = Fernet.generate_key()
with open(self.key_file, 'wb') as f:
f.write(self.key)
# 设置文件权限（仅所有者可读）
os.chmod(self.key_file, 0o400)
self.cipher = Fernet(self.key)
def encrypt(self, plaintext: str) -> str:
"""加密文本"""
encrypted = self.cipher.encrypt(plaintext.encode())
return b64encode(encrypted).decode()
def decrypt(self, encrypted_text: str) -> str:
"""解密文本"""
encrypted = b64decode(encrypted_text.encode())
return self.cipher.decrypt(encrypted).decode()
# 使用环境变量和加密
class SecureConfig:
"""安全配置管理"""
@staticmethod
def get_secret(key, default=None, required=False):
"""获取安全密钥"""
value = os.environ.get(key)
if value is None:
if required:
raise ValueError(f"必须设置环境变量: {key}")
return default
# 如果是加密值，进行解密
if value.startswith('ENC:'):
try:
manager = SecretManager()
return manager.decrypt(value[4:])
except Exception as e:
raise ValueError(f"解密失败: {key} - {e}")
return value
@classmethod
def load_secure_config(cls, app):
"""加载安全配置"""
# 数据库配置
app.config['SQLALCHEMY_DATABASE_URI'] = cls.get_secret(
'DATABASE_URL', required=True
)
# API密钥
app.config['STRIPE_SECRET_KEY'] = cls.get_secret(
'STRIPE_SECRET_KEY', required=True
)
# 邮件密码
app.config['MAIL_PASSWORD'] = cls.get_secret(
'MAIL_PASSWORD', required=True
)
# 其他可选配置
app.config['SENTRY_DSN'] = cls.get_secret('SENTRY_DSN')
app.config['AWS_SECRET_ACCESS_KEY'] = cls.get_secret('AWS_SECRET_ACCESS_KEY')
# 在生产环境中使用
# 设置环境变量（已加密）
# export DATABASE_URL="ENC:gAAAAABf..."
# export STRIPE_SECRET_KEY="ENC:gAAAAABf..."# config/development.py
import os
from .base import Config
class DevelopmentConfig(Config):
"""开发环境配置"""
# 基础
ENV = 'development'
DEBUG = True
TESTING = False
# 数据库
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///dev.db'
SQLALCHEMY_ECHO = True # 打印SQL语句
# 安全（开发环境放宽限制）
SESSION_COOKIE_SECURE = False
WTF_CSRF_ENABLED = False # 开发时可禁用CSRF
# 调试工具
DEBUG_TB_ENABLED = True
DEBUG_TB_INTERCEPT_REDIRECTS = False
DEBUG_TB_PANELS = [
'flask_debugtoolbar.panels.versions.VersionDebugPanel',
'flask_debugtoolbar.panels.timer.TimerDebugPanel',
'flask_debugtoolbar.panels.headers.HeaderDebugPanel',
'flask_debugtoolbar.panels.request_vars.RequestVarsDebugPanel',
'flask_debugtoolbar.panels.template.TemplateDebugPanel',
'flask_debugtoolbar.panels.sqlalchemy.SQLAlchemyDebugPanel',
'flask_debugtoolbar.panels.logger.LoggingPanel',
'flask_debugtoolbar.panels.profiler.ProfilerDebugPanel',
]
# 热重载
TEMPLATES_AUTO_RELOAD = True
EXPLAIN_TEMPLATE_LOADING = True
# 缓存控制
SEND_FILE_MAX_AGE_DEFAULT = 0 # 禁用缓存
# 邮件
MAIL_SUPPRESS_SEND = True # 不实际发送邮件
MAIL_DEBUG = True
# 日志
LOG_LEVEL = 'DEBUG'
LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
@classmethod
def init_app(cls, app):
"""开发环境初始化"""
super().init_app(app)
# 启用调试工具栏
if app.config.get('DEBUG_TB_ENABLED'):
try:
from flask_debugtoolbar import DebugToolbarExtension
toolbar = DebugToolbarExtension(app)
except ImportError:
app.logger.warning('Flask-DebugToolbar未安装')
# 详细的请求日志
@app.before_request
def log_request_info():
app.logger.debug('Headers: %s', dict(request.headers))
app.logger.debug('Body: %s', request.get_data())# config/production.py
import os
import logging
from logging.handlers import RotatingFileHandler, SMTPHandler
from .base import Config
class ProductionConfig(Config):
"""生产环境配置"""
# 基础
ENV = 'production'
DEBUG = False
TESTING = False
# 必须从环境变量读取
SECRET_KEY = os.environ['SECRET_KEY']
SQLALCHEMY_DATABASE_URI = os.environ['DATABASE_URL']
# 安全增强
SESSION_COOKIE_SECURE = True
REMEMBER_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
PREFERRED_URL_SCHEME = 'https'
# 数据库连接池
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 20,
'pool_recycle': 3600,
'pool_pre_ping': True,
'max_overflow': 10,
'pool_timeout': 30,
}
# 性能优化
JSON_SORT_KEYS = False
JSONIFY_PRETTYPRINT_REGULAR = False
SEND_FILE_MAX_AGE_DEFAULT = 31536000 # 1年缓存
# 邮件
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = True
MAIL_USERNAME = os.environ['MAIL_USERNAME']
MAIL_PASSWORD = os.environ['MAIL_PASSWORD']
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER', MAIL_USERNAME)
# 日志
LOG_LEVEL = 'WARNING'
LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s [%(request_id)s]'
# 限流
RATELIMIT_ENABLED = True
RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL')
RATELIMIT_DEFAULT = ["200 per day", "50 per hour"]
RATELIMIT_STRATEGY = 'fixed-window'
# 错误追踪
SENTRY_DSN = os.environ.get('SENTRY_DSN')
# CDN配置
CDN_DOMAIN = os.environ.get('CDN_DOMAIN')
@classmethod
def init_app(cls, app):
"""生产环境初始化"""
super().init_app(app)
# 配置日志
cls.setup_logging(app)
# 配置监控
cls.setup_monitoring(app)
# 配置健康检查
cls.setup_health_checks(app)
# 配置安全头
cls.setup_security_headers(app)
@staticmethod
def setup_logging(app):
"""配置生产环境日志"""
# 创建日志目录
log_dir = '/var/log/flask'
os.makedirs(log_dir, exist_ok=True)
# 文件处理器（按大小轮转）
file_handler = RotatingFileHandler(
os.path.join(log_dir, 'app.log'),
maxBytes=10*1024*1024, # 10MB
backupCount=10
)
file_handler.setLevel(logging.WARNING)
# 错误日志单独文件
error_handler = RotatingFileHandler(
os.path.join(log_dir, 'error.log'),
maxBytes=10*1024*1024,
backupCount=10
)
error_handler.setLevel(logging.ERROR)
# 格式化器
formatter = logging.Formatter(app.config['LOG_FORMAT'])
file_handler.setFormatter(formatter)
error_handler.setFormatter(formatter)
# 添加处理器
app.logger.addHandler(file_handler)
app.logger.addHandler(error_handler)
# 邮件错误通知
if app.config.get('ADMIN_EMAIL'):
mail_handler = SMTPHandler(
mailhost=(app.config['MAIL_SERVER'], app.config['MAIL_PORT']),
fromaddr=app.config['MAIL_DEFAULT_SENDER'],
toaddrs=[app.config['ADMIN_EMAIL']],
subject='应用错误 - {}'.format(app.config['APP_NAME']),
credentials=(app.config['MAIL_USERNAME'], app.config['MAIL_PASSWORD']),
secure=() if app.config.get('MAIL_USE_TLS') else None
)
mail_handler.setLevel(logging.ERROR)
mail_handler.setFormatter(formatter)
app.logger.addHandler(mail_handler)
# 设置日志级别
app.logger.setLevel(app.config['LOG_LEVEL'])
# 禁用Flask默认的日志处理器
logging.getLogger('werkzeug').setLevel(logging.WARNING)
@staticmethod
def setup_monitoring(app):
"""配置监控"""
# Sentry错误监控
if app.config.get('SENTRY_DSN'):
try:
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
sentry_sdk.init(
dsn=app.config['SENTRY_DSN'],
integrations=[FlaskIntegration()],
environment=os.environ.get('SENTRY_ENVIRONMENT', 'production'),
release=app.config.get('VERSION', '1.0.0'),
traces_sample_rate=0.1,
)
app.logger.info('Sentry监控已启用')
except ImportError:
app.logger.warning('Sentry未安装，跳过监控配置')
# New Relic监控
if os.environ.get('NEW_RELIC_LICENSE_KEY'):
try:
import newrelic.agent
newrelic.agent.initialize('newrelic.ini')
app.logger.info('New Relic监控已启用')
except ImportError:
app.logger.warning('New Relic未安装，跳过监控配置')
@staticmethod
def setup_health_checks(app):
"""配置健康检查端点"""
@app.route('/health')
def health_check():
return {'status': 'healthy', 'timestamp': datetime.now().isoformat()}
@app.route('/metrics')
@login_required
@admin_required
def metrics():
"""应用指标端点"""
import psutil
import json
metrics = {
'memory': psutil.virtual_memory()._asdict(),
'cpu': psutil.cpu_percent(interval=1),
'disk': psutil.disk_usage('/')._asdict(),
'uptime': time.time() - psutil.boot_time(),
}
return json.dumps(metrics)
@staticmethod
def setup_security_headers(app):
"""配置安全HTTP头"""
@app.after_request
def add_security_headers(response):
# 安全头
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-XSS-Protection'] = '1; mode=block'
response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
# CSP头（根据实际需求调整）
if not app.debug:
response.headers['Content-Security-Policy'] = \
"default-src 'self'; " \
"script-src 'self' https://cdn.jsdelivr.net; " \
"style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; " \
"img-src 'self' data: https:; " \
"font-src 'self' https://cdn.jsdelivr.net;"
return response# config/validators.py
import os
import re
from typing import Dict, Any, List, Optional
from urllib.parse import urlparse
class ConfigValidator:
"""配置验证器"""
@staticmethod
def validate_database_url(url: str) -> bool:
"""验证数据库URL"""
try:
parsed = urlparse(url)
# 检查协议
if parsed.scheme not in ['postgresql', 'mysql', 'sqlite']:
return False
# 检查必要组件
if parsed.scheme != 'sqlite' and not parsed.netloc:
return False
return True
except:
return False
@staticmethod
def validate_secret_key(key: str) -> bool:
"""验证密钥强度"""
if len(key) < 16:
return False
# 检查是否包含足够的不同字符类型
has_upper = any(c.isupper() for c in key)
has_lower = any(c.islower() for c in key)
has_digit = any(c.isdigit() for c in key)
has_special = any(not c.isalnum() for c in key)
# 至少包含3种字符类型
types_count = sum([has_upper, has_lower, has_digit, has_special])
return types_count >= 3
@staticmethod
def validate_email(email: str) -> bool:
"""验证邮箱格式"""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
@staticmethod
def validate_url(url: str) -> bool:
"""验证URL格式"""
try:
parsed = urlparse(url)
return all([parsed.scheme, parsed.netloc])
except:
return False
@staticmethod
def validate_port(port: Any) -> bool:
"""验证端口号"""
try:
port_int = int(port)
return 1 <= port_int <= 65535
except:
return False
@classmethod
def validate_config(cls, config: Dict[str, Any]) -> List[str]:
"""验证完整配置"""
errors = []
# 必需配置项
required_fields = [
('SECRET_KEY', str, cls.validate_secret_key, "必须设置强密码的SECRET_KEY"),
('SQLALCHEMY_DATABASE_URI', str, cls.validate_database_url, "无效的数据库URL"),
]
for field, expected_type, validator, error_msg in required_fields:
if field not in config:
errors.append(f"缺少必需配置: {field}")
continue
value = config[field]
# 类型检查
if not isinstance(value, expected_type):
errors.append(f"{field} 应该是 {expected_type.__name__} 类型")
continue
# 自定义验证
if validator and not validator(value):
errors.append(f"{field}: {error_msg}")
# 可选配置验证
if 'MAIL_DEFAULT_SENDER' in config:
if not cls.validate_email(config['MAIL_DEFAULT_SENDER']):
errors.append("MAIL_DEFAULT_SENDER: 无效的邮箱格式")
if 'SERVER_NAME' in config:
if not cls.validate_url(f"http://{config['SERVER_NAME']}"):
errors.append("SERVER_NAME: 无效的服务器名称")
if 'MAIL_PORT' in config:
if not cls.validate_port(config['MAIL_PORT']):
errors.append("MAIL_PORT: 无效的端口号")
return errors
# 在应用启动时验证配置
def validate_app_config(app):
"""验证应用配置"""
validator = ConfigValidator()
errors = validator.validate_config(app.config)
if errors:
error_msg = "\n".join(errors)
if app.debug:
# 开发环境：直接抛出异常
raise ValueError(f"配置验证失败:\n{error_msg}")
else:
# 生产环境：记录错误并退出
app.logger.error(f"配置验证失败:\n{error_msg}")
# 发送警报邮件
send_alert_email("配置验证失败", error_msg)
# 优雅退出
import sys
sys.exit(1)
app.logger.info("配置验证通过")#!/usr/bin/env python
# check_config.py - 配置检查脚本
import os
import sys
import argparse
from dotenv import load_dotenv
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
def check_environment():
"""检查环境变量"""
print("🔍 检查环境变量...")
required_vars = [
('SECRET_KEY', True),
('DATABASE_URL', True),
('MAIL_USERNAME', False),
('MAIL_PASSWORD', False),
('REDIS_URL', False),
]
missing = []
for var, required in required_vars:
if var in os.environ:
if var in ['SECRET_KEY', 'MAIL_PASSWORD']:
print(f" ✓ {var}: 已设置 (值隐藏)")
else:
print(f" ✓ {var}: {os.environ[var][:50]}...")
elif required:
missing.append(var)
print(f" ✗ {var}: 未设置 (必需)")
else:
print(f" ⚠ {var}: 未设置 (可选)")
return missing
def check_database():
"""检查数据库连接"""
print("\n🔍 检查数据库连接...")
try:
from config import Config
config = Config()
if not hasattr(config, 'SQLALCHEMY_DATABASE_URI'):
print(" ✗ 未配置数据库URL")
return False
from sqlalchemy import create_engine
engine = create_engine(config.SQLALCHEMY_DATABASE_URI)
with engine.connect() as conn:
conn.execute("SELECT 1")
print(" ✓ 数据库连接成功")
return True
except Exception as e:
print(f" ✗ 数据库连接失败: {e}")
return False
def check_redis():
"""检查Redis连接"""
print("\n🔍 检查Redis连接...")
try:
import redis
redis_url = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
r = redis.from_url(redis_url)
r.ping()
print(" ✓ Redis连接成功")
return True
except Exception as e:
print(f" ⚠ Redis连接失败: {e}")
return False
def check_security():
"""检查安全配置"""
print("\n🔍 检查安全配置...")
from config.validators import ConfigValidator
validator = ConfigValidator()
# 检查SECRET_KEY强度
secret_key = os.environ.get('SECRET_KEY')
if secret_key:
if validator.validate_secret_key(secret_key):
print(" ✓ SECRET_KEY强度足够")
else:
print(" ⚠ SECRET_KEY强度不足，建议使用更复杂的密钥")
# 检查调试模式
if os.environ.get('FLASK_DEBUG') == '1':
print(" ⚠ 调试模式已启用，生产环境请禁用")
# 检查HTTPS设置
if os.environ.get('SESSION_COOKIE_SECURE') != 'True':
print(" ⚠ SESSION_COOKIE_SECURE未启用，生产环境建议启用")
def main():
"""主函数"""
parser = argparse.ArgumentParser(description='检查项目配置')
parser.add_argument('--env', help='环境变量文件路径', default='.env')
args = parser.parse_args()
# 加载环境变量
if os.path.exists(args.env):
load_dotenv(args.env)
print(f"📁 已加载环境变量文件: {args.env}")
else:
print(f"📁 未找到环境变量文件: {args.env}，使用系统环境变量")
print("=" * 50)
print("🛠 开始配置检查")
print("=" * 50)
# 执行检查
missing_vars = check_environment()
db_ok = check_database()
redis_ok = check_redis()
check_security()
print("\n" + "=" * 50)
print("📊 检查结果汇总")
print("=" * 50)
# 统计结果
errors = []
warnings = []
if missing_vars:
errors.append(f"缺少必需环境变量: {', '.join(missing_vars)}")
if not db_ok:
errors.append("数据库连接失败")
if not redis_ok:
warnings.append("Redis连接失败")
# 输出结果
if errors:
print("❌ 发现错误:")
for error in errors:
print(f" • {error}")
if warnings:
print("⚠️ 发现警告:")
for warning in warnings:
print(f" • {warning}")
if not errors and not warnings:
print("✅ 所有检查通过，配置正常！")
print("=" * 50)
return 0 if not errors else 1
if __name__ == '__main__':
sys.exit(main())# Dockerfile
# 多阶段构建
# 第一阶段：构建阶段
FROM python:3.9-slim as builder
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
g++ \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
# 创建虚拟环境并安装依赖
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
RUN pip install --no-cache-dir -r requirements.txt
# 第二阶段：运行阶段
FROM python:3.9-slim
# 安装运行时依赖
RUN apt-get update && apt-get install -y \
libpq5 \
&& rm -rf /var/lib/apt/lists/*
# 创建非root用户
RUN groupadd -r flask && useradd -r -g flask flask
# 创建应用目录
WORKDIR /app
# 从构建阶段复制虚拟环境
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# 复制应用代码
COPY . .
# 设置权限
RUN chown -R flask:flask /app
USER flask
# 创建必要的目录
RUN mkdir -p /app/logs /app/uploads
# 环境变量
ENV FLASK_APP=app.py
ENV FLASK_ENV=production
ENV PYTHONUNBUFFERED=1
ENV PYTHONPATH=/app
# 暴露端口
EXPOSE 5000
# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD curl -f http://localhost:5000/health || exit 1
# 启动命令
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "app:create_app()"]# docker-compose.prod.yml
version: '3.8'
services:
web:
build:
context: .
dockerfile: Dockerfile
image: myapp:${TAG:-latest}
container_name: myapp-web
restart: unless-stopped
depends_on:
- db
- redis
environment:
- FLASK_ENV=production
- DATABASE_URL=postgresql://${DB_USER}:${DB_PASSWORD}@db:5432/${DB_NAME}
- REDIS_URL=redis://redis:6379/0
- SECRET_KEY=${SECRET_KEY}
- MAIL_USERNAME=${MAIL_USERNAME}
- MAIL_PASSWORD=${MAIL_PASSWORD}
env_file:
- .env.production
volumes:
- uploads:/app/uploads
- logs:/app/logs
ports:
- "${HOST_PORT}:5000"
networks:
- app-network
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
db:
image: postgres:13-alpine
container_name: myapp-db
restart: unless-stopped
environment:
POSTGRES_DB: ${DB_NAME}
POSTGRES_USER: ${DB_USER}
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
- postgres_data:/var/lib/postgresql/data
- ./docker/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
ports:
- "5432:5432"
networks:
- app-network
command: postgres -c max_connections=100 -c shared_buffers=256MB
redis:
image: redis:6-alpine
container_name: myapp-redis
restart: unless-stopped
command: redis-server --appendonly yes
volumes:
- redis_data:/data
ports:
- "6379:6379"
networks:
- app-network
nginx:
image: nginx:alpine
container_name: myapp-nginx
restart: unless-stopped
depends_on:
- web
ports:
- "80:80"
- "443:443"
volumes:
- ./docker/nginx/nginx.conf:/etc/nginx/nginx.conf
- ./docker/nginx/ssl:/etc/nginx/ssl
- ./logs/nginx:/var/log/nginx
networks:
- app-network
volumes:
postgres_data:
redis_data:
uploads:
logs:
networks:
app-network:
driver: bridge# k8s/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: flask-app-config
data:
# 应用配置
FLASK_ENV: "production"
APP_NAME: "MyFlaskApp"
LOG_LEVEL: "INFO"
# 数据库配置（敏感信息使用Secret）
DB_HOST: "postgres-service"
DB_PORT: "5432"
DB_NAME: "production_db"
# Redis配置
REDIS_HOST: "redis-service"
REDIS_PORT: "6379"
# 邮件配置
MAIL_SERVER: "smtp.gmail.com"
MAIL_PORT: "587"
MAIL_USE_TLS: "true"
# k8s/secret.yaml
apiVersion: v1
kind: Secret
metadata:
name: flask-app-secrets
type: Opaque
stringData:
# 安全密钥
SECRET_KEY: "production-secret-key-change-this"
# 数据库凭据
DB_USER: "flask_user"
DB_PASSWORD: "production-db-password"
# 邮件凭据
MAIL_USERNAME: "noreply@example.com"
MAIL_PASSWORD: "mail-app-password"
# API密钥
STRIPE_SECRET_KEY: "sk_live_..."
GOOGLE_MAPS_API_KEY: "AIzaSy..."
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: flask-app
labels:
app: flask-app
spec:
replicas: 3
selector:
matchLabels:
app: flask-app
template:
metadata:
labels:
app: flask-app
spec:
containers:
- name: flask-app
image: myregistry.com/myapp:latest
ports:
- containerPort: 5000
env:
- name: SECRET_KEY
valueFrom:
secretKeyRef:
name: flask-app-secrets
key: SECRET_KEY
- name: DATABASE_URL
value: "postgresql://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):$(DB_PORT)/$(DB_NAME)"
envFrom:
- configMapRef:
name: flask-app-config
- secretRef:
name: flask-app-secrets
resources:
requests:
memory: "256Mi"
cpu: "250m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 5000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /health
port: 5000
initialDelaySeconds: 5
periodSeconds: 5推荐的多环境配置管理策略：
# 示例：多环境配置管理
import os
# 环境检测
ENV = os.environ.get('FLASK_ENV', 'development')
# 配置文件映射
CONFIG_FILES = {
'development': 'config/development.py',
'testing': 'config/testing.py',
'production': 'config/production.py',
'staging': 'config/staging.py',
}
def load_config(app):
"""根据环境加载配置"""
# 1. 从环境变量读取配置名
config_name = ENV
# 2. 加载对应配置文件
config_file = CONFIG_FILES.get(config_name)
if config_file and os.path.exists(config_file):
app.config.from_pyfile(config_file)
# 3. 环境变量覆盖（最高优先级）
app.config.from_prefixed_env()
# 4. 验证配置
validate_config(app.config)
# 或者使用工厂模式
class ConfigFactory:
@staticmethod
def create_config(env=None):
if env is None:
env = os.environ.get('FLASK_ENV', 'development')
configs = {
'dev': DevelopmentConfig,
'test': TestingConfig,
'prod': ProductionConfig,
'staging': StagingConfig,
}
return configs.get(env, DevelopmentConfig)()敏感信息的安全管理：
| 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 环境变量 | 简单，广泛支持 | 管理复杂，可能泄露 | 小型到中型项目 |
| 加密配置文件 | 可版本控制 | 需要密钥管理 | 需要共享配置 |
| 密钥管理服务 | 安全，集中管理 | 复杂，有成本 | 企业级应用 |
| Docker Secrets | Docker原生支持 | 仅限Docker环境 | 容器化部署 |
# 使用AWS Secrets Manager
import boto3
import json
def get_secret(secret_name):
"""从AWS Secrets Manager获取密钥"""
client = boto3.client('secretsmanager')
try:
response = client.get_secret_value(SecretId=secret_name)
secret = json.loads(response['SecretString'])
return secret
except Exception as e:
print(f"获取密钥失败: {e}")
return None
# 在配置中使用
class AwsConfig:
@classmethod
def load_secrets(cls, app):
"""从AWS加载密钥"""
secrets = get_secret('myapp/production')
if secrets:
app.config['DATABASE_URL'] = secrets['database_url']
app.config['SECRET_KEY'] = secrets['secret_key']
app.config['STRIPE_SECRET_KEY'] = secrets['stripe_secret_key']
# 使用HashiCorp Vault
import hvac
def get_vault_secret(path):
"""从Vault获取密钥"""
client = hvac.Client(
url=os.environ['VAULT_ADDR'],
token=os.environ['VAULT_TOKEN']
)
secret = client.read(path)
return secret['data']配置热重载的实现方案：
# 配置热重载实现
import os
import threading
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class ConfigReloader:
"""配置热重载器"""
def __init__(self, app, config_file):
self.app = app
self.config_file = config_file
self.last_modified = os.path.getmtime(config_file)
self.observer = Observer()
self.running = False
def start(self):
"""启动配置监视"""
event_handler = FileSystemEventHandler()
event_handler.on_modified = self.on_config_modified
self.observer.schedule(
event_handler,
path=os.path.dirname(self.config_file),
recursive=False
)
self.observer.start()
self.running = True
# 启动后台线程
thread = threading.Thread(target=self.monitor_loop, daemon=True)
thread.start()
def stop(self):
"""停止配置监视"""
self.running = False
self.observer.stop()
self.observer.join()
def on_config_modified(self, event):
"""配置文件修改事件处理"""
if event.src_path == self.config_file:
current_modified = os.path.getmtime(self.config_file)
# 避免重复触发
if current_modified > self.last_modified + 1:
self.last_modified = current_modified
self.reload_config()
def reload_config(self):
"""重新加载配置"""
try:
self.app.logger.info("重新加载配置文件...")
# 保存当前配置状态
old_debug = self.app.config.get('DEBUG')
old_secret = self.app.config.get('SECRET_KEY')
# 重新加载配置
self.app.config.from_pyfile(self.config_file, silent=True)
# 检查敏感配置是否改变
new_secret = self.app.config.get('SECRET_KEY')
if new_secret != old_secret:
self.app.logger.warning("SECRET_KEY已改变，需要重启应用")
# 这里可以触发重启逻辑
self.app.logger.info("配置重新加载完成")
# 触发配置变更事件
self.app.extensions.get('config_signals', {}).get(
'config_changed', lambda: None
)()
except Exception as e:
self.app.logger.error(f"配置重新加载失败: {e}")
def monitor_loop(self):
"""监视循环（用于环境变量等）"""
while self.running:
# 检查环境变量变化
self.check_env_vars()
time.sleep(60) # 每分钟检查一次
def check_env_vars(self):
"""检查环境变量变化"""
env_vars_to_watch = [
'DATABASE_URL', 'REDIS_URL', 'MAIL_PASSWORD'
]
for var in env_vars_to_watch:
old_value = getattr(self, f'last_{var}', None)
new_value = os.environ.get(var)
if new_value != old_value:
setattr(self, f'last_{var}', new_value)
self.app.logger.info(f"环境变量 {var} 已改变")
# 更新应用配置
if var in self.app.config:
self.app.config[var] = new_value
# 在应用中使用
app = Flask(__name__)
# 初始化配置重载器
if app.debug: # 只在开发环境启用
reloader = ConfigReloader(app, 'config/development.py')
reloader.start()
# 注册配置变更信号
config_signals = {}
app.extensions['config_signals'] = config_signals
@app.route('/config/reload')
@admin_required
def reload_config():
"""手动触发配置重载"""
reloader.reload_config()
return "配置已重新加载"配置版本控制策略：
# 配置版本管理
import json
import hashlib
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Dict, Any
@dataclass
class ConfigVersion:
"""配置版本"""
version: int
timestamp: datetime
hash: str
author: str
comment: str
config: Dict[str, Any]
class ConfigVersionControl:
"""配置版本控制"""
def __init__(self, storage_path='config/versions'):
self.storage_path = storage_path
os.makedirs(storage_path, exist_ok=True)
def save_version(self, config: Dict[str, Any], author: str, comment: str = ""):
"""保存配置版本"""
# 获取当前版本号
versions = self.list_versions()
version = len(versions) + 1
# 计算配置哈希
config_json = json.dumps(config, sort_keys=True, default=str)
config_hash = hashlib.sha256(config_json.encode()).hexdigest()
# 创建版本对象
config_version = ConfigVersion(
version=version,
timestamp=datetime.now(),
hash=config_hash,
author=author,
comment=comment,
config=config
)
# 保存到文件
filename = f"config_v{version:03d}_{config_hash[:8]}.json"
filepath = os.path.join(self.storage_path, filename)
with open(filepath, 'w') as f:
json.dump(asdict(config_version), f, indent=2, default=str)
# 更新最新版本链接
latest_link = os.path.join(self.storage_path, 'latest.json')
with open(latest_link, 'w') as f:
json.dump({'version': version, 'file': filename}, f)
return config_version
def list_versions(self) -> list[ConfigVersion]:
"""列出所有配置版本"""
versions = []
for filename in os.listdir(self.storage_path):
if filename.startswith('config_v') and filename.endswith('.json'):
filepath = os.path.join(self.storage_path, filename)
with open(filepath, 'r') as f:
data = json.load(f)
# 转换时间戳
data['timestamp'] = datetime.fromisoformat(data['timestamp'])
versions.append(ConfigVersion(**data))
# 按版本号排序
versions.sort(key=lambda x: x.version)
return versions
def get_version(self, version: int) -> ConfigVersion:
"""获取特定版本配置"""
for config_version in self.list_versions():
if config_version.version == version:
return config_version
raise ValueError(f"版本 {version} 不存在")
def diff_versions(self, version1: int, version2: int) -> Dict[str, Any]:
"""比较两个版本的差异"""
v1 = self.get_version(version1)
v2 = self.get_version(version2)
diff = {}
# 比较所有配置项
all_keys = set(v1.config.keys()) | set(v2.config.keys())
for key in all_keys:
val1 = v1.config.get(key)
val2 = v2.config.get(key)
if val1 != val2:
diff[key] = {
'from': val1,
'to': val2,
'changed': True
}
return diff
def rollback(self, version: int, app):
"""回滚到指定版本"""
config_version = self.get_version(version)
# 应用配置
for key, value in config_version.config.items():
app.config[key] = value
# 保存回滚记录
self.save_version(
config=dict(app.config),
author='system',
comment=f'回滚到版本 {version}'
)
return config_version
# 使用示例
version_control = ConfigVersionControl()
# 保存配置版本
@app.before_first_request
def save_initial_config():
"""保存初始配置版本"""
version_control.save_version(
config=dict(app.config),
author='system',
comment='初始配置'
)
# 配置变更API
@app.route('/api/config/versions')
@admin_required
def list_config_versions():
"""列出配置版本"""
versions = version_control.list_versions()
return jsonify([asdict(v) for v in versions])
@app.route('/api/config/diff/<int:v1>/<int:v2>')
@admin_required
def diff_config_versions(v1, v2):
"""比较配置版本差异"""
diff = version_control.diff_versions(v1, v2)
return jsonify(diff)
@app.route('/api/config/rollback/<int:version>', methods=['POST'])
@admin_required
def rollback_config(version):
"""回滚配置"""
version_control.rollback(version, app)
return jsonify({'message': f'已回滚到版本 {version}'})