qidao123.com技术社区-IT企服评测·应用市场

标题: [flask]中心件 [打印本页]
作者: 卖不甜枣    时间: 2025-5-4 22:30
标题: [flask]中心件
前言

中心件用于在哀求到达视图函数之前或响应返回给客户端之前执行一些操作，比如身份验证、日志记录、错误处理等。
方式1，使用钩子函数
  1. from flask import Flask, request, Response
  3. def DemoMiddleWare(app: Flask):
  4.     @app.before_request
  5.     def before_request1():
  6.         print("before_request")
  8.     @app.after_request
  9.     def after_request1(response: Response):
  10.         print(f"after_request, method: {request.method}, path: {request.path}")
  11.         return response
  12.    
  13.     return app
  15. app = Flask(__name__)
  17. @app.get("/")
  18. def hello():
  19.     return "Hello World!"
  21. app = DemoMiddleWare(app)
  23. if __name__ == "__main__":
  24.     app.run(debug=False)
复制代码
方式2，实现中心件类
  1. from flask import Flask
  2. from werkzeug.middleware.proxy_fix import ProxyFix
  3. import time
  6. class MyMiddleWare:
  7.     def __init__(self, app):
  8.         self.app = app
  10.     def __call__(self, environ, start_response):
  11.         data = {
  12.             "method": environ.get("REQUEST_METHOD", "NaN"),
  13.             "scheme": environ.get("wsgi.url_scheme", "NaN"),
  14.             "server_protocol": environ.get("SERVER_PROTOCOL", "NaN"),
  15.             "remote_addr": environ.get("REMOTE_ADDR", "NaN"),
  16.             "host": environ.get("HTTP_HOST", "NaN"),
  17.             "path": environ.get("PATH_INFO", "NaN"),
  18.             "user_agent": environ.get("HTTP_USER_AGENT", "NaN"),
  19.         }
  21.         # 自定义请求头都以 HTTP_* 开头
  22.         if header_ha := environ.get("HTTP_HA", ""):
  23.             data["header_ha"] = header_ha
  25.         # 处理 cookie
  26.         if cookie := environ.get("HTTP_COOKIE", ""):
  27.             data["cookie"] = cookie
  29.         start_time = time.time()
  31.         response = self.app(environ, start_response)
  32.         data["response_time"] = round((time.time() - start_time), 4)
  33.         print(data)
  35.         return response
  38. class My2MiddleWare:
  39.     def __init__(self, app):
  40.         self.app = app
  42.     def __call__(self, environ, start_response):
  43.         print("This is middleware2, before request")
  44.         response = self.app(environ, start_response)
  45.         print("This is middleware2, after request")
  47.         return response
  50. app = Flask(__name__)
  51. # ProxyFix是werkzeug提供的一个中间件，当flask应用部署在反向代理器后面时，可以将反向代理器设置得X-Forwarded-*头信息
  52. # 具体ProxyFix处理了哪些代理请求头可以参考ProxyFix的docstring
  53. app.wsgi_app = ProxyFix(app.wsgi_app)
  54. app.wsgi_app = MyMiddleWare(app.wsgi_app)
  55. app.wsgi_app = My2MiddleWare(app.wsgi_app)  # 添加第2个中间件
  58. @app.get("/")
  59. def hello():
  60.     return "Hello World!"
  63. @app.get("/2")
  64. def hello2():
  65.     return "Hello World!"
  68. if __name__ == "__main__":
  69.     app.run(debug=False)
复制代码
上面示例代码只处理了哀求头等信息，偶然候可能还需要处理响应，比如获取响应码、设置响应头等。
  1. from flask import Flask
  2. import time
  5. class MyMiddleWare:
  6.     def __init__(self, app):
  7.         self.app = app
  9.     def __call__(self, environ, start_response):
  10.         data = {
  11.             "method": environ.get("REQUEST_METHOD", "NaN"),
  12.             "scheme": environ.get("wsgi.url_scheme", "NaN"),
  13.             "server_protocol": environ.get("SERVER_PROTOCOL", "NaN"),
  14.             "remote_addr": environ.get("REMOTE_ADDR", "NaN"),
  15.             "host": environ.get("HTTP_HOST", "NaN"),
  16.             "path": environ.get("PATH_INFO", "NaN"),
  17.             "user_agent": environ.get("HTTP_USER_AGENT", "NaN"),
  18.         }
  20.         # 自定义请求头都以 HTTP_* 开头
  21.         if header_ha := environ.get("HTTP_HA", ""):
  22.             data["header_ha"] = header_ha
  24.         # 处理 cookie
  25.         if cookie := environ.get("HTTP_COOKIE", ""):
  26.             data["cookie"] = cookie
  28.         resp_status_code = None
  29.         resp_headers = []
  31.         def catching_start_response(status, headers, exc_info=None):
  32.             nonlocal resp_status_code, resp_headers
  33.             resp_status_code = status
  34.             headers.append(("X-Resp1", "OK"))  # 添加自定义响应头
  35.             resp_headers = headers
  36.             return start_response(status, headers, exc_info)
  38.         start_time = time.time()
  39.         response = self.app(environ, catching_start_response)
  40.         data["status"] = resp_status_code
  41.         data["response_time"] = round((time.time() - start_time), 4)
  42.         print(data)
  43.         print(f"response headers: {resp_headers}")
  45.         return response
  48. app = Flask(__name__)
  49. app.wsgi_app = MyMiddleWare(app.wsgi_app)
  52. @app.get("/")
  53. def hello():
  54.     return "Hello World!"
  57. @app.get("/2")
  58. def hello2():
  59.     return "Hello World!"
  62. if __name__ == "__main__":
  63.     app.run(debug=False)
复制代码
小结

两种方式实现中心件都行，不外用钩子函数会更方便点。假如想水代码行数，用中心件类也是不错的选择。
参考


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/) Powered by Discuz! X3.4