[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)

复制代码

小结

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

• Flask入门（九）：Flask中心件
  
• Flask官方文档
  

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