HoRain云--Flask中间件与扩展全解析

HoRain云小助手个人主页 个人专栏: 《Linux 系列教程》《c语言教程》⛺️生活的理想就是为了理想的生活!⛳️ 推荐前些天发现了一个超棒的服务器购买网站性价比超高大内存超划算忍不住分享一下给大家。点击跳转到网站。专栏介绍专栏名称专栏介绍《C语言》本专栏主要撰写C干货内容和编程技巧让大家从底层了解C把更多的知识由抽象到简单通俗易懂。《网络协议》本专栏主要是注重从底层来给大家一步步剖析网络协议的奥秘一起解密网络协议在运行中协议的基本运行机制《docker容器精解篇》全面深入解析 docker 容器从基础到进阶涵盖原理、操作、实践案例助您精通 docker。《linux系列》本专栏主要撰写Linux干货内容从基础到进阶知识由抽象到简单通俗易懂帮你从新手小白到扫地僧。《python 系列》本专栏着重撰写Python相关的干货内容与编程技巧助力大家从底层去认识Python将更多复杂的知识由抽象转化为简单易懂的内容。《试题库》本专栏主要是发布一些考试和练习题库涵盖软考、HCIE、HRCE、CCNA等目录⛳️ 推荐专栏介绍Flask 中间件和扩展1. Flask 中间件1.1 基本概念1.2 常用的 WSGI 中间件1.2.1 ProxyFix处理反向代理1.2.2 自定义中间件1.3 更完整的中间件示例1.4 Flask 的 before/after 请求钩子2. Flask 扩展2.1 常用扩展列表2.2 主要扩展详细用法2.2.1 Flask-SQLAlchemy2.2.2 Flask-Migrate2.2.3 Flask-WTF2.2.4 Flask-Login2.2.5 Flask-JWT-Extended2.2.6 Flask-CORS2.2.7 Flask-Caching2.2.8 Flask-Limiter3. 自定义扩展3.1 创建简单的扩展3.2 使用自定义扩展4. 扩展开发最佳实践4.1 扩展结构4.2 完整的扩展示例5. 最佳实践总结Flask 中间件和扩展1. Flask 中间件Flask 中间件本质上是 WSGI 中间件因为 Flask 应用是一个 WSGI 应用。1.1 基本概念from flask import Flask, request, g import time app Flask(__name__)1.2 常用的 WSGI 中间件1.2.1 ProxyFix处理反向代理from werkzeug.middleware.proxy_fix import ProxyFix # 配置在反向代理后的应用 app.wsgi_app ProxyFix( app.wsgi_app, x_for1, # 信任的 X-Forwarded-For 头数量 x_proto1, # 信任的 X-Forwarded-Proto 头数量 x_host1, # 信任的 X-Forwarded-Host 头数量 x_port1, # 信任的 X-Forwarded-Port 头数量 x_prefix1 # 信任的 X-Forwarded-Prefix 头数量 )1.2.2 自定义中间件class TimingMiddleware: 记录请求处理时间的中间件 def __init__(self, app): self.app app def __call__(self, environ, start_response): start_time time.time() def custom_start_response(status, headers, exc_infoNone): # 计算处理时间 process_time time.time() - start_time headers.append((X-Process-Time, str(process_time))) return start_response(status, headers, exc_info) return self.app(environ, custom_start_response) # 使用中间件 app.wsgi_app TimingMiddleware(app.wsgi_app)1.3 更完整的中间件示例class LoggingMiddleware: 日志记录中间件 def __init__(self, app, loggerNone): self.app app self.logger logger or logging.getLogger(__name__) def __call__(self, environ, start_response): # 请求前记录 request_start_time time.time() request_method environ.get(REQUEST_METHOD) request_path environ.get(PATH_INFO) remote_addr environ.get(REMOTE_ADDR) self.logger.info(fRequest started: {request_method} {request_path} from {remote_addr}) def custom_start_response(status, headers, exc_infoNone): # 请求后记录 process_time time.time() - request_start_time status_code status.split( )[0] self.logger.info( fRequest completed: {status_code} f{request_method} {request_path} fin {process_time:.3f}s ) return start_response(status, headers, exc_info) return self.app(environ, custom_start_response) class AuthMiddleware: 认证中间件 def __init__(self, app, api_key_headerX-API-Key): self.app app self.api_key_header api_key_header self.valid_keys {your-api-key-here, another-valid-key} def __call__(self, environ, start_response): # 跳过某些路径的认证 path environ.get(PATH_INFO, ) if path.startswith(/public/): return self.app(environ, start_response) # 检查 API Key api_key environ.get(fHTTP_{self.api_key_header.upper().replace(-, _)}) if not api_key or api_key not in self.valid_keys: # 认证失败 start_response(401 Unauthorized, [(Content-Type, text/plain)]) return [bUnauthorized] return self.app(environ, start_response) # 组合使用多个中间件 app.wsgi_app LoggingMiddleware(app.wsgi_app) app.wsgi_app AuthMiddleware(app.wsgi_app)1.4 Flask 的 before/after 请求钩子虽然不是标准的 WSGI 中间件但提供了类似的功能from flask import Flask, g, request, jsonify import time app Flask(__name__) app.before_request def before_request(): 在每次请求前执行 g.start_time time.time() g.request_id request.headers.get(X-Request-ID, unknown) # 记录请求开始 app.logger.info(fRequest {g.request_id} started: {request.method} {request.path}) # 检查 API 密钥简化示例 if request.endpoint and not request.endpoint.startswith(static): api_key request.headers.get(X-API-Key) if not is_valid_api_key(api_key): return jsonify({error: Invalid API key}), 401 app.after_request def after_request(response): 在每次请求后执行 # 计算处理时间 if hasattr(g, start_time): process_time time.time() - g.start_time response.headers[X-Process-Time] str(process_time) app.logger.info(fRequest {g.request_id} completed in {process_time:.3f}s) # 添加 CORS 头 response.headers[Access-Control-Allow-Origin] * response.headers[Access-Control-Allow-Headers] Content-Type, Authorization response.headers[Access-Control-Allow-Methods] GET, POST, PUT, DELETE, OPTIONS return response app.teardown_request def teardown_request(exceptionNone): 请求结束后执行用于清理资源 if hasattr(g, db_session): g.db_session.remove() if exception: app.logger.error(fRequest {g.request_id} failed: {exception}) def is_valid_api_key(api_key): 验证 API 密钥 valid_keys {test-key-123, prod-key-456} return api_key in valid_keys2. Flask 扩展2.1 常用扩展列表扩展用途安装命令Flask-SQLAlchemyORM 数据库集成pip install flask-sqlalchemyFlask-Migrate数据库迁移pip install flask-migrateFlask-WTF表单处理pip install flask-wtfFlask-Login用户会话管理pip install flask-loginFlask-RESTfulREST API 支持pip install flask-restfulFlask-JWT-ExtendedJWT 认证pip install flask-jwt-extendedFlask-CORS跨域资源共享pip install flask-corsFlask-Caching缓存支持pip install flask-cachingFlask-SocketIOWebSocket 支持pip install flask-socketioFlask-Mail邮件发送pip install flask-mailFlask-APScheduler定时任务pip install flask-apschedulerFlask-Limiter请求限流pip install flask-limiter2.2 主要扩展详细用法2.2.1 Flask-SQLAlchemyfrom flask import Flask from flask_sqlalchemy import SQLAlchemy from datetime import datetime app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] sqlite:///app.db app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app) # 定义模型 class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) email db.Column(db.String(120), uniqueTrue, nullableFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow) posts db.relationship(Post, backrefauthor, lazyTrue) def __repr__(self): return fUser {self.username} class Post(db.Model): id db.Column(db.Integer, primary_keyTrue) title db.Column(db.String(120), nullableFalse) content db.Column(db.Text, nullableFalse) user_id db.Column(db.Integer, db.ForeignKey(user.id), nullableFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow) # 使用 app.route(/users) def get_users(): users User.query.all() return {users: [user.username for user in users]} # 创建数据库表 with app.app_context(): db.create_all()2.2.2 Flask-Migratefrom flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] sqlite:///app.db db SQLAlchemy(app) migrate Migrate(app, db) # 使用命令 # flask db init # 初始化迁移目录 # flask db migrate # 生成迁移脚本 # flask db upgrade # 应用迁移 # flask db downgrade # 回滚迁移2.2.3 Flask-WTFfrom flask import Flask, render_template, flash, redirect, url_for from flask_wtf import FlaskForm from wtforms import StringField, PasswordField, TextAreaField, SubmitField from wtforms.validators import DataRequired, Email, Length, EqualTo from wtforms.validators import ValidationError app Flask(__name__) app.config[SECRET_KEY] your-secret-key-here # 自定义验证器 def password_strength(form, field): password field.data if len(password) 8: raise ValidationError(密码至少需要8个字符) if not any(c.isdigit() for c in password): raise ValidationError(密码必须包含数字) # 表单定义 class RegistrationForm(FlaskForm): username StringField(用户名, validators[ DataRequired(message用户名不能为空), Length(min3, max20, message用户名长度必须在3-20个字符之间) ]) email StringField(邮箱, validators[ DataRequired(message邮箱不能为空), Email(message请输入有效的邮箱地址) ]) password PasswordField(密码, validators[ DataRequired(message密码不能为空), password_strength ]) confirm_password PasswordField(确认密码, validators[ DataRequired(message请确认密码), EqualTo(password, message两次输入的密码不一致) ]) submit SubmitField(注册) # 路由中使用 app.route(/register, methods[GET, POST]) def register(): form RegistrationForm() if form.validate_on_submit(): # 处理表单提交 flash(注册成功, success) return redirect(url_for(index)) return render_template(register.html, formform)2.2.4 Flask-Loginfrom flask import Flask, render_template, redirect, url_for, flash from flask_login import ( LoginManager, UserMixin, login_user, logout_user, login_required, current_user ) from werkzeug.security import generate_password_hash, check_password_hash app Flask(__name__) app.config[SECRET_KEY] your-secret-key-here login_manager LoginManager() login_manager.init_app(app) login_manager.login_view login login_manager.login_message 请先登录 login_manager.login_message_category warning # 用户模型 class User(UserMixin): def __init__(self, id, username, password_hash): self.id id self.username username self.password_hash password_hash def check_password(self, password): return check_password_hash(self.password_hash, password) # 模拟用户数据库 users { 1: User(1, admin, generate_password_hash(admin123)), 2: User(2, user, generate_password_hash(user123)) } login_manager.user_loader def load_user(user_id): return users.get(int(user_id)) app.route(/login, methods[GET, POST]) def login(): if current_user.is_authenticated: return redirect(url_for(dashboard)) if request.method POST: username request.form.get(username) password request.form.get(password) # 查找用户 user next((u for u in users.values() if u.username username), None) if user and user.check_password(password): login_user(user, rememberTrue) flash(登录成功, success) return redirect(url_for(dashboard)) else: flash(用户名或密码错误, error) return render_template(login.html) app.route(/logout) login_required def logout(): logout_user() flash(已退出登录, info) return redirect(url_for(login)) app.route(/dashboard) login_required def dashboard(): return render_template(dashboard.html, usercurrent_user)2.2.5 Flask-JWT-Extendedfrom flask import Flask, jsonify, request from flask_jwt_extended import ( JWTManager, create_access_token, create_refresh_token, jwt_required, get_jwt_identity, get_jwt ) from datetime import timedelta from werkzeug.security import generate_password_hash, check_password_hash app Flask(__name__) app.config[JWT_SECRET_KEY] super-secret-key-change-this app.config[JWT_ACCESS_TOKEN_EXPIRES] timedelta(hours1) app.config[JWT_REFRESH_TOKEN_EXPIRES] timedelta(days30) jwt JWTManager(app) # 模拟数据库 users_db { admin: { password_hash: generate_password_hash(admin123), role: admin }, user: { password_hash: generate_password_hash(user123), role: user } } app.route(/login, methods[POST]) def login(): username request.json.get(username) password request.json.get(password) user users_db.get(username) if not user or not check_password_hash(user[password_hash], password): return jsonify({error: 用户名或密码错误}), 401 # 创建令牌 access_token create_access_token( identityusername, additional_claims{role: user[role]} ) refresh_token create_refresh_token(identityusername) return jsonify({ access_token: access_token, refresh_token: refresh_token }) app.route(/protected, methods[GET]) jwt_required() def protected(): current_user get_jwt_identity() claims get_jwt() return jsonify({ message: f你好, {current_user}!, role: claims.get(role), user: current_user }) app.route(/refresh, methods[POST]) jwt_required(refreshTrue) def refresh(): identity get_jwt_identity() access_token create_access_token(identityidentity) return jsonify({access_token: access_token}) jwt.expired_token_loader def expired_token_callback(jwt_header, jwt_payload): return jsonify({ error: 令牌已过期, message: 请重新登录 }), 401 jwt.invalid_token_loader def invalid_token_callback(error): return jsonify({ error: 无效令牌, message: 请提供有效的令牌 }), 4012.2.6 Flask-CORSfrom flask import Flask, jsonify from flask_cors import CORS app Flask(__name__) # 简单用法允许所有来源 # CORS(app) # 高级配置 CORS(app, resources{ r/api/*: { origins: [http://localhost:3000, https://example.com], methods: [GET, POST, PUT, DELETE, OPTIONS], allow_headers: [Content-Type, Authorization], expose_headers: [X-Custom-Header], supports_credentials: True, max_age: 3600 } }) app.route(/api/data) def get_data(): return jsonify({data: 跨域数据}) # 或者针对特定路由 from flask_cors import cross_origin app.route(/api/users) cross_origin( origins[http://localhost:3000], methods[GET, POST], allow_headers[Content-Type] ) def get_users(): return jsonify({users: [Alice, Bob, Charlie]})2.2.7 Flask-Cachingfrom flask import Flask from flask_caching import Cache import time app Flask(__name__) # 配置缓存 app.config[CACHE_TYPE] simple # 可以使用 redis, memcached 等 app.config[CACHE_DEFAULT_TIMEOUT] 300 # 默认缓存时间秒 cache Cache(app) # 使用装饰器缓存视图 app.route(/expensive-operation) cache.cached(timeout50) # 缓存50秒 def expensive_operation(): time.sleep(2) # 模拟耗时操作 return {result: expensive data} # 缓存带参数的视图 app.route(/user/int:user_id) cache.memoize(timeout60) # 根据参数缓存 def get_user(user_id): time.sleep(1) return {user_id: user_id, name: fUser {user_id}} # 手动缓存 app.route(/manual-cache) def manual_cache(): data cache.get(my_key) if data is None: data calculate_expensive_data() cache.set(my_key, data, timeout60) return data def calculate_expensive_data(): time.sleep(3) return {data: 计算结果}2.2.8 Flask-Limiterfrom flask import Flask, jsonify from flask_limiter import Limiter from flask_limiter.util import get_remote_address app Flask(__name__) limiter Limiter( appapp, key_funcget_remote_address, # 默认使用客户端IP default_limits[200 per day, 50 per hour] ) # 全局限制 app.route(/api/data) limiter.limit(10 per minute) def get_data(): return jsonify({data: some data}) # 不同限制策略 app.route(/api/public) limiter.limit(100/hour) def public_api(): return jsonify({message: 公共API}) app.route(/api/premium) limiter.limit(1000/hour) def premium_api(): return jsonify({message: 高级API}) # 动态限制 app.route(/api/dynamic) limiter.limit(lambda: 10/minute if is_weekend() else 100/minute) def dynamic_limit(): return jsonify({message: 动态限制}) def is_weekend(): import datetime return datetime.datetime.today().weekday() 5 # 多个限制条件 app.route(/api/complex) limiter.limit(100/day;10/minute;5/second) def complex_limit(): return jsonify({message: 复杂限制}) # 错误处理 limiter.request_filter def header_whitelist(): # 白名单过滤器 return request.headers.get(X-Forwarded-For, ).startswith(192.168.) app.errorhandler(429) def ratelimit_handler(e): return jsonify({ error: 请求过于频繁, message: str(e.description) }), 4293. 自定义扩展3.1 创建简单的扩展 my_extension/__init__.py 自定义 Flask 扩展示例 from flask import current_app, g import json class MyExtension: 一个简单的 Flask 扩展 def __init__(self, appNone): self.app app if app is not None: self.init_app(app) def init_app(self, app): 初始化扩展 # 默认配置 app.config.setdefault(MY_EXTENSION_SETTING, default_value) # 在应用上下文中执行初始化 with app.app_context(): self._setup_extensions() # 注册关闭时的清理函数 app.teardown_appcontext(self.teardown) # 将扩展实例存储在 app.extensions 中 app.extensions[my_extension] self # 添加模板全局变量 app.context_processor def inject_my_extension(): return {my_extension: self} def _setup_extensions(self): 设置扩展内部组件 self._cache {} def teardown(self, exception): 清理资源 if hasattr(g, my_extension_session): g.my_extension_session.close() def do_something(self, data): 扩展的功能方法 app self.app or current_app app.logger.debug(fMyExtension processing: {data}) return fProcessed: {data} def cache_data(self, key, value): 缓存数据 self._cache[key] value def get_cached_data(self, key): 获取缓存数据 return self._cache.get(key) # 工厂模式创建扩展实例 my_ext MyExtension()3.2 使用自定义扩展from flask import Flask, jsonify, g from my_extension import MyExtension app Flask(__name__) app.config[MY_EXTENSION_SETTING] custom_value # 初始化扩展 my_ext MyExtension(app) # 或者my_ext.init_app(app) app.route(/) def index(): # 通过 current_app 访问扩展 result current_app.extensions[my_extension].do_something(test) return jsonify({result: result}) app.route(/cache/key) def cache_data(key): # 使用扩展功能 my_ext current_app.extensions[my_extension] # 先从缓存获取 cached my_ext.get_cached_data(key) if cached: return jsonify({data: cached, cached: True}) # 缓存数据 data {key: key, value: some value} my_ext.cache_data(key, data) return jsonify({data: data, cached: False})4. 扩展开发最佳实践4.1 扩展结构my_flask_extension/ ├── setup.py ├── README.md ├── LICENSE ├── my_flask_extension/ │ ├── __init__.py │ ├── core.py │ ├── utils.py │ └── static/ │ └── extension.css └── tests/ └── test_extension.py4.2 完整的扩展示例 一个完整的 Flask 扩展示例 import time from functools import wraps from flask import current_app, request, g, has_request_context class TimingExtension: 一个记录请求处理时间的扩展 def __init__(self, appNone): self.app app self._timings {} if app is not None: self.init_app(app) def init_app(self, app): 初始化应用 # 默认配置 app.config.setdefault(TIMING_ENABLED, True) app.config.setdefault(TIMING_THRESHOLD, 1.0) # 秒 # 注册请求钩子 app.before_request(self._before_request) app.after_request(self._after_request) app.teardown_request(self._teardown_request) # 添加到扩展字典 if not hasattr(app, extensions): app.extensions {} app.extensions[timing] self def _before_request(self): 请求开始前 if not current_app.config[TIMING_ENABLED]: return g.request_start_time time.time() g.request_id id(request) # 记录开始时间 self._timings[g.request_id] { start: g.request_start_time, end: None, duration: None, path: request.path, method: request.method } def _after_request(self, response): 请求结束后 if not current_app.config[TIMING_ENABLED]: return response if hasattr(g, request_id) and g.request_id in self._timings: request_data self._timings[g.request_id] request_data[end] time.time() request_data[duration] request_data[end] - request_data[start] # 添加响应头 if request_data[duration] current_app.config[TIMING_THRESHOLD]: response.headers[X-Slow-Request] true response.headers[X-Request-Duration] str(request_data[duration]) return response def _teardown_request(self, exceptionNone): 清理请求上下文 if hasattr(g, request_id): # 可以选择保留或删除计时数据 pass def get_timings(self): 获取所有计时数据 return self._timings.copy() def clear_timings(self): 清除所有计时数据 self._timings.clear() def timer(self, name): 装饰器记录函数执行时间 def decorator(f): wraps(f) def decorated_function(*args, **kwargs): if not has_request_context(): return f(*args, **kwargs) if not hasattr(g, function_timers): g.function_timers {} start_time time.time() try: result f(*args, **kwargs) finally: end_time time.time() duration end_time - start_time if name not in g.function_timers: g.function_timers[name] { count: 0, total_time: 0.0 } g.function_timers[name][count] 1 g.function_timers[name][total_time] duration return result return decorated_function return decorator # 创建扩展实例 timing TimingExtension()5. 最佳实践总结中间件选择使用 Werkzeug 中间件处理 WSGI 层功能使用 Flask 钩子处理应用层逻辑扩展管理使用工厂模式创建应用和扩展合理配置扩展顺序注意扩展之间的依赖关系性能考虑合理使用缓存扩展监控慢请求实现请求限流安全性使用 JWT 或 Flask-Login 进行认证配置合适的 CORS 策略验证和清理用户输入错误处理为扩展添加适当的错误处理提供有意义的错误消息实现优雅降级测试为扩展编写单元测试测试扩展与 Flask 的集成模拟各种使用场景通过合理使用中间件和扩展可以显著提高 Flask 应用的开发效率和运行性能同时保持代码的整洁和可维护性。❤️❤️❤️本人水平有限如有纰漏欢迎各位大佬评论批评指正如果觉得这篇文对你有帮助的话也请给个点赞、收藏下吧非常感谢! Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧