⚡ 3. Flask 核心概念
🔌 3.1 WSGI 协议与应用生命周期
WSGI 协议深入理解
WSGI(Web Server Gateway Interface)是 Python Web 应用与 Web 服务器之间的标准接口。Flask 作为 WSGI 应用,需要深入理解这个协议才能更好地掌握其工作原理。
WSGI 应用结构
python
# 最简单的 WSGI 应用
def simple_wsgi_app(environ, start_response):
"""最基础的 WSGI 应用示例"""
status = '200 OK'
headers = [('Content-Type', 'text/plain; charset=utf-8')]
start_response(status, headers)
return [b'Hello, WSGI World!']
# Flask 应用的 WSGI 接口
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return 'Hello, Flask!'
# Flask 应用本身就是一个 WSGI 应用
# app.__call__ 方法实现了 WSGI 接口
if __name__ == '__main__':
from wsgiref.simple_server import make_server
server = make_server('localhost', 8000, app)
server.serve_forever()1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Flask 应用生命周期
应用生命周期钩子
python
from flask import Flask, g, request
import time
app = Flask(__name__)
# 应用启动前
@app.before_first_request
def before_first_request():
"""在处理第一个请求前执行"""
print("应用首次启动,执行初始化操作")
# 可以在这里进行数据库连接、缓存预热等操作
# 每个请求前
@app.before_request
def before_request():
"""在每个请求处理前执行"""
g.start_time = time.time()
g.user_id = request.headers.get('X-User-ID')
print(f"请求开始: {request.method} {request.path}")
# 每个请求后
@app.after_request
def after_request(response):
"""在每个请求处理后执行"""
duration = time.time() - g.start_time
print(f"请求完成,耗时: {duration:.3f}s")
response.headers['X-Response-Time'] = str(duration)
return response
# 请求结束时(无论是否有异常)
@app.teardown_request
def teardown_request(exception):
"""请求结束时执行清理工作"""
if exception:
print(f"请求异常: {exception}")
# 清理资源,如数据库连接等
# 应用上下文结束时
@app.teardown_appcontext
def teardown_appcontext(exception):
"""应用上下文结束时执行"""
# 清理应用级别的资源
pass1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
🏭 3.2 应用工厂模式(Application Factory)
传统应用结构的问题
python
# 传统方式 - 存在问题
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
# 全局变量,难以测试和配置
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)
@app.route('/')
def index():
return 'Hello World'
# 问题:
# 1. 难以进行单元测试
# 2. 无法动态配置
# 3. 循环导入问题
# 4. 扩展初始化时机固定1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
应用工厂模式实现
python
# app/__init__.py - 应用工厂
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_migrate import Migrate
# 创建扩展实例(但不初始化)
db = SQLAlchemy()
login_manager = LoginManager()
migrate = Migrate()
def create_app(config_name='development'):
"""应用工厂函数"""
app = Flask(__name__)
# 加载配置
app.config.from_object(f'config.{config_name.title()}Config')
# 初始化扩展
db.init_app(app)
login_manager.init_app(app)
migrate.init_app(app, db)
# 配置登录管理器
login_manager.login_view = 'auth.login'
login_manager.login_message = '请先登录访问此页面'
# 注册蓝图
from app.main import bp as main_bp
app.register_blueprint(main_bp)
from app.auth import bp as auth_bp
app.register_blueprint(auth_bp, url_prefix='/auth')
from app.api import bp as api_bp
app.register_blueprint(api_bp, url_prefix='/api')
# 注册错误处理器
register_error_handlers(app)
# 注册CLI命令
register_cli_commands(app)
return app
def register_error_handlers(app):
"""注册错误处理器"""
@app.errorhandler(404)
def not_found_error(error):
return render_template('errors/404.html'), 404
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return render_template('errors/500.html'), 500
def register_cli_commands(app):
"""注册CLI命令"""
@app.cli.command()
def init_db():
"""初始化数据库"""
db.create_all()
print('数据库初始化完成')1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
配置管理
python
# config.py - 配置管理
import os
from datetime import timedelta
class Config:
"""基础配置类"""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
SQLALCHEMY_TRACK_MODIFICATIONS = False
PERMANENT_SESSION_LIFETIME = timedelta(days=7)
# 邮件配置
MAIL_SERVER = os.environ.get('MAIL_SERVER')
MAIL_PORT = int(os.environ.get('MAIL_PORT') or 587)
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
# 分页配置
POSTS_PER_PAGE = 10
USERS_PER_PAGE = 20
@staticmethod
def init_app(app):
"""应用特定的初始化"""
pass
class DevelopmentConfig(Config):
"""开发环境配置"""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///' + os.path.join(os.path.dirname(__file__), 'data-dev.sqlite')
# 开发环境特定配置
SQLALCHEMY_ECHO = True # 显示SQL查询
WTF_CSRF_TIME_LIMIT = None # 禁用CSRF超时
class TestingConfig(Config):
"""测试环境配置"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
"""生产环境配置"""
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(os.path.dirname(__file__), 'data.sqlite')
@classmethod
def init_app(cls, app):
Config.init_app(app)
# 生产环境特定初始化
import logging
from logging.handlers import RotatingFileHandler
if not app.debug:
if not os.path.exists('logs'):
os.mkdir('logs')
file_handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('应用启动')
# 配置字典
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
应用启动文件
python
# run.py - 应用启动
import os
from app import create_app, db
from app.models import User, Post
# 从环境变量获取配置,默认为开发环境
app = create_app(os.getenv('FLASK_CONFIG') or 'development')
@app.shell_context_processor
def make_shell_context():
"""为 flask shell 命令提供上下文"""
return {'db': db, 'User': User, 'Post': Post}
if __name__ == '__main__':
app.run(debug=True)1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
2
3
4
5
6
7
8
9
10
11
12
13
14
15
🛣️ 3.3 路由系统深入解析
路由注册机制
路由装饰器深入
python
from flask import Flask, request, url_for
from functools import wraps
app = Flask(__name__)
# 基础路由
@app.route('/')
def index():
return 'Hello World'
# 多种HTTP方法
@app.route('/api/users', methods=['GET', 'POST'])
def users():
if request.method == 'GET':
return {'users': []}
elif request.method == 'POST':
return {'message': 'User created'}, 201
# 动态路由参数
@app.route('/user/<int:user_id>')
def show_user(user_id):
return f'User ID: {user_id}'
@app.route('/post/<slug>')
def show_post(slug):
return f'Post: {slug}'
# 可选参数
@app.route('/page/')
@app.route('/page/<int:page>')
def show_page(page=1):
return f'Page: {page}'
# 路径参数(包含斜杠)
@app.route('/path/<path:subpath>')
def show_subpath(subpath):
return f'Subpath: {subpath}'
# UUID参数
@app.route('/object/<uuid:object_id>')
def show_object(object_id):
return f'Object ID: {object_id}'1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
自定义路由转换器
python
from werkzeug.routing import BaseConverter
class ListConverter(BaseConverter):
"""自定义列表转换器"""
def to_python(self, value):
return value.split(',')
def to_url(self, values):
return ','.join(BaseConverter.to_url(value) for value in values)
class RegexConverter(BaseConverter):
"""正则表达式转换器"""
def __init__(self, url_map, *items):
super(RegexConverter, self).__init__(url_map)
self.regex = items[0]
# 注册自定义转换器
app.url_map.converters['list'] = ListConverter
app.url_map.converters['regex'] = RegexConverter
# 使用自定义转换器
@app.route('/tags/<list:tags>')
def show_tags(tags):
return f'Tags: {tags}'
@app.route('/user/<regex("[a-z]+"):username>')
def show_user_regex(username):
return f'Username: {username}'1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
路由组织与蓝图
python
# app/main/routes.py - 主要路由
from flask import Blueprint, render_template, request, current_app
from app.models import Post
bp = Blueprint('main', __name__)
@bp.route('/')
@bp.route('/index')
def index():
page = request.args.get('page', 1, type=int)
posts = Post.query.paginate(
page=page,
per_page=current_app.config['POSTS_PER_PAGE'],
error_out=False
)
return render_template('index.html', posts=posts)
@bp.route('/about')
def about():
return render_template('about.html')
# app/auth/routes.py - 认证路由
from flask import Blueprint, render_template, redirect, url_for, flash
from flask_login import login_user, logout_user, current_user
from app.auth.forms import LoginForm, RegistrationForm
from app.models import User
from app import db
bp = Blueprint('auth', __name__)
@bp.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user and user.check_password(form.password.data):
login_user(user, remember=form.remember_me.data)
return redirect(url_for('main.index'))
flash('用户名或密码错误')
return render_template('auth/login.html', form=form)
@bp.route('/logout')
def logout():
logout_user()
return redirect(url_for('main.index'))1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
🔄 3.4 请求和响应处理
请求对象详解
python
from flask import Flask, request, jsonify
import json
app = Flask(__name__)
@app.route('/request-info', methods=['GET', 'POST', 'PUT', 'DELETE'])
def request_info():
"""展示请求对象的各种属性"""
info = {
# 基本信息
'method': request.method,
'url': request.url,
'base_url': request.base_url,
'url_root': request.url_root,
'path': request.path,
'query_string': request.query_string.decode(),
# 请求头
'headers': dict(request.headers),
'user_agent': str(request.user_agent),
'remote_addr': request.remote_addr,
# 参数
'args': dict(request.args), # GET参数
'form': dict(request.form), # POST表单数据
'files': list(request.files.keys()), # 上传文件
# JSON数据
'is_json': request.is_json,
'json': request.get_json() if request.is_json else None,
# 其他
'cookies': dict(request.cookies),
'endpoint': request.endpoint,
'view_args': request.view_args,
}
return jsonify(info)
# 处理不同类型的请求数据
@app.route('/data', methods=['POST'])
def handle_data():
"""处理不同格式的请求数据"""
# JSON数据
if request.is_json:
data = request.get_json()
return jsonify({'received': data, 'type': 'json'})
# 表单数据
elif request.form:
data = dict(request.form)
return jsonify({'received': data, 'type': 'form'})
# 原始数据
elif request.data:
data = request.data.decode('utf-8')
return jsonify({'received': data, 'type': 'raw'})
else:
return jsonify({'error': 'No data received'}), 4001
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
响应对象构建
python
from flask import Flask, make_response, jsonify, render_template, redirect, url_for
from datetime import datetime, timedelta
app = Flask(__name__)
# 简单字符串响应
@app.route('/simple')
def simple_response():
return 'Hello World'
# 带状态码的响应
@app.route('/with-status')
def response_with_status():
return 'Created', 201
# 带头部的响应
@app.route('/with-headers')
def response_with_headers():
return 'Hello', 200, {'X-Custom-Header': 'Custom Value'}
# 使用 make_response 构建复杂响应
@app.route('/complex')
def complex_response():
resp = make_response('Complex Response')
resp.status_code = 200
resp.headers['X-Custom'] = 'Value'
resp.headers['Cache-Control'] = 'no-cache'
# 设置Cookie
resp.set_cookie('session_id', 'abc123',
expires=datetime.now() + timedelta(days=30),
secure=True, httponly=True)
return resp
# JSON响应
@app.route('/json')
def json_response():
data = {
'message': 'Success',
'timestamp': datetime.now().isoformat(),
'data': [1, 2, 3, 4, 5]
}
return jsonify(data)
# 文件下载响应
@app.route('/download')
def download_file():
from flask import send_file
return send_file('static/files/document.pdf',
as_attachment=True,
download_name='my_document.pdf')
# 重定向响应
@app.route('/redirect-example')
def redirect_example():
return redirect(url_for('simple_response'))
# 流式响应
@app.route('/stream')
def stream_response():
def generate():
for i in range(1000):
yield f'data chunk {i}\n'
return app.response_class(generate(), mimetype='text/plain')1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
响应处理中间件
python
@app.after_request
def after_request(response):
"""响应后处理中间件"""
# 添加安全头
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-XSS-Protection'] = '1; mode=block'
# CORS处理
if request.method == 'OPTIONS':
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE'
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
# 性能监控
if hasattr(g, 'start_time'):
response.headers['X-Response-Time'] = str(time.time() - g.start_time)
return response1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
🔗 3.5 中间件与钩子函数
请求处理流程
中间件实现模式
python
from flask import Flask, request, g, session
from functools import wraps
import time
import logging
app = Flask(__name__)
app.secret_key = 'your-secret-key'
# 性能监控中间件
class PerformanceMiddleware:
def __init__(self, app):
self.app = app
self.init_app(app)
def init_app(self, app):
app.before_request(self.before_request)
app.after_request(self.after_request)
def before_request(self):
g.start_time = time.time()
def after_request(self, response):
duration = time.time() - g.start_time
if duration > 1.0: # 记录慢请求
app.logger.warning(f'Slow request: {request.path} took {duration:.3f}s')
response.headers['X-Response-Time'] = f'{duration:.3f}'
return response
# 认证中间件
class AuthMiddleware:
def __init__(self, app):
self.app = app
self.init_app(app)
def init_app(self, app):
app.before_request(self.load_user)
def load_user(self):
user_id = session.get('user_id')
if user_id:
# 从数据库加载用户信息
g.current_user = User.query.get(user_id)
else:
g.current_user = None
# 注册中间件
performance = PerformanceMiddleware(app)
auth = AuthMiddleware(app)1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
装饰器形式的中间件
python
def require_auth(f):
"""认证装饰器"""
@wraps(f)
def decorated_function(*args, **kwargs):
if not g.current_user:
return jsonify({'error': 'Authentication required'}), 401
return f(*args, **kwargs)
return decorated_function
def require_role(role):
"""角色权限装饰器"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if not g.current_user or g.current_user.role != role:
return jsonify({'error': 'Insufficient permissions'}), 403
return f(*args, **kwargs)
return decorated_function
return decorator
def rate_limit(max_requests=100, window=3600):
"""限流装饰器"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
# 实现限流逻辑
client_ip = request.remote_addr
# 检查请求频率...
return f(*args, **kwargs)
return decorated_function
return decorator
# 使用装饰器
@app.route('/admin/users')
@require_auth
@require_role('admin')
@rate_limit(max_requests=50, window=3600)
def admin_users():
return jsonify({'users': []})1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
⚙️ 3.6 配置管理与环境变量
配置层次结构
高级配置管理
python
# config.py - 高级配置管理
import os
from datetime import timedelta
from typing import Type
class Config:
"""基础配置类"""
# 基础配置
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
# 数据库配置
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_RECORD_QUERIES = True
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_pre_ping': True,
'pool_recycle': 300,
}
# 会话配置
PERMANENT_SESSION_LIFETIME = timedelta(days=7)
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
# 安全配置
WTF_CSRF_TIME_LIMIT = 3600
BCRYPT_LOG_ROUNDS = 12
# 邮件配置
MAIL_SERVER = os.environ.get('MAIL_SERVER')
MAIL_PORT = int(os.environ.get('MAIL_PORT') or 587)
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_SUBJECT_PREFIX = '[MyApp] '
MAIL_SENDER = 'MyApp Admin <admin@myapp.com>'
# Redis配置
REDIS_URL = os.environ.get('REDIS_URL') or 'redis://localhost:6379/0'
# 缓存配置
CACHE_TYPE = 'redis'
CACHE_REDIS_URL = REDIS_URL
CACHE_DEFAULT_TIMEOUT = 300
# 文件上传配置
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER = os.path.join(os.path.dirname(__file__), 'uploads')
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}
# API配置
API_TITLE = 'My API'
API_VERSION = 'v1'
OPENAPI_VERSION = '3.0.2'
# 分页配置
POSTS_PER_PAGE = 10
USERS_PER_PAGE = 20
COMMENTS_PER_PAGE = 50
@staticmethod
def init_app(app):
"""应用特定的初始化"""
pass
@classmethod
def get_config(cls) -> dict:
"""获取所有配置项"""
return {key: getattr(cls, key) for key in dir(cls)
if not key.startswith('_') and key.isupper()}
class DevelopmentConfig(Config):
"""开发环境配置"""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///' + os.path.join(os.path.dirname(__file__), 'data-dev.sqlite')
# 开发环境特定配置
SQLALCHEMY_ECHO = True
WTF_CSRF_ENABLED = False # 开发时禁用CSRF
SESSION_COOKIE_SECURE = False # HTTP环境
# 邮件配置(开发环境使用控制台输出)
MAIL_SUPPRESS_SEND = True
@classmethod
def init_app(cls, app):
Config.init_app(app)
# 开发环境日志配置
import logging
logging.basicConfig(level=logging.DEBUG)
class TestingConfig(Config):
"""测试环境配置"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
SESSION_COOKIE_SECURE = False
# 测试环境特定配置
BCRYPT_LOG_ROUNDS = 4 # 加快测试速度
MAIL_SUPPRESS_SEND = True
class ProductionConfig(Config):
"""生产环境配置"""
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(os.path.dirname(__file__), 'data.sqlite')
@classmethod
def init_app(cls, app):
Config.init_app(app)
# 生产环境日志配置
import logging
from logging.handlers import RotatingFileHandler, SMTPHandler
# 文件日志
if not os.path.exists('logs'):
os.mkdir('logs')
file_handler = RotatingFileHandler(
'logs/app.log', maxBytes=10240000, backupCount=10)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
# 邮件日志(错误通知)
if cls.MAIL_SERVER:
auth = None
if cls.MAIL_USERNAME or cls.MAIL_PASSWORD:
auth = (cls.MAIL_USERNAME, cls.MAIL_PASSWORD)
secure = None
if cls.MAIL_USE_TLS:
secure = ()
mail_handler = SMTPHandler(
mailhost=(cls.MAIL_SERVER, cls.MAIL_PORT),
fromaddr=cls.MAIL_SENDER,
toaddrs=['admin@myapp.com'],
subject='MyApp Failure',
credentials=auth, secure=secure)
mail_handler.setLevel(logging.ERROR)
app.logger.addHandler(mail_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('MyApp startup')
# 配置字典
config: dict[str, Type[Config]] = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
环境变量管理
python
# .env 文件示例
FLASK_APP=run.py
FLASK_ENV=development
SECRET_KEY=your-secret-key-here
DATABASE_URL=postgresql://user:pass@localhost/myapp
REDIS_URL=redis://localhost:6379/0
MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USE_TLS=true
MAIL_USERNAME=your-email@gmail.com
MAIL_PASSWORD=your-app-password
# 环境变量加载
from dotenv import load_dotenv
import os
# 加载.env文件
load_dotenv()
# 获取环境变量的辅助函数
def get_env_variable(var_name: str, default=None, var_type=str):
"""获取环境变量并进行类型转换"""
value = os.environ.get(var_name, default)
if value is None:
return None
if var_type == bool:
return value.lower() in ['true', '1', 'on', 'yes']
elif var_type == int:
return int(value)
elif var_type == float:
return float(value)
else:
return value
# 使用示例
class Config:
SECRET_KEY = get_env_variable('SECRET_KEY', 'dev-secret-key')
DEBUG = get_env_variable('DEBUG', False, bool)
DATABASE_URL = get_env_variable('DATABASE_URL')
REDIS_URL = get_env_variable('REDIS_URL', 'redis://localhost:6379/0')
MAIL_PORT = get_env_variable('MAIL_PORT', 587, int)1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
通过以上深入的 Flask 核心概念讲解,您将全面理解 Flask 的工作原理和核心机制,为构建复杂的 Web 应用奠定坚实的理论基础。
