Python3中Sanic中間件的使用
在 Sanic 中，中間件（middleware）是指在請求和響應之間執(zhí)行的代碼。它們是一個非常強大的工具，用于處理請求的預處理、響應的后處理、全局錯誤處理、日志記錄、認證、權限校驗、跨域資源共享（CORS）等任務。中間件通常用于應用程序的全局邏輯，在請求處理前或響應處理后執(zhí)行。
Sanic 中間件的工作流程
Sanic 中間件會在以下兩個階段執(zhí)行：
- 請求處理前：處理請求數(shù)據(jù)，在請求被路由處理之前執(zhí)行?？梢杂糜谛ｒ?、修改請求等操作。
- 響應處理后：處理響應數(shù)據(jù)，在響應被發(fā)送給客戶端之前執(zhí)行?？梢杂糜谛薷捻憫⒂涗浫罩镜炔僮?。
中間件的使用
在 Sanic 中，中間件通過裝飾器來定義。你可以為整個應用程序（全局中間件）定義中間件，也可以為某些特定的路由定義中間件。
1. 全局中間件
全局中間件是在整個應用程序的生命周期中被調用的，不依賴于特定的路由。它們在請求和響應的各個階段都會執(zhí)行。
請求階段的全局中間件
全局中間件通常用于處理請求的預處理，如身份驗證、記錄日志、設置 CORS 等。
from sanic import Sanic
from sanic.response import json
app = Sanic("MyApp")
# 請求階段的全局中間件
@app.middleware("request")
async def log_request(request):
print(f"Request received: {request.method} {request.url}")
# 你可以在這里對請求進行修改，例如身份驗證、設置 CORS 等
# 如果請求需要認證，可以在這里進行驗證
# 響應階段的全局中間件
@app.middleware("response")
async def log_response(request, response):
print(f"Response status: {response.status}")
# 你可以在這里修改響應，例如添加 CORS 頭部、記錄日志等
# 在響應返回給客戶端之前處理
response.headers['X-Custom-Header'] = 'Sanic' # 可以修改響應頭
return response
@app.route("/")
async def hello(request):
return json({"message": "Hello, Sanic!"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
在這個例子中：
log_request中間件會在每個請求到達應用時被調用，打印請求的方法和 URL。log_response中間件會在每個響應返回給客戶端之前被調用，打印響應的狀態(tài)碼，并在響應頭中添加一個自定義的 header。
2. 路由中間件
你還可以為特定的路由或路由組定義中間件。這是 Sanic 提供的靈活性，可以讓你為某些路由添加額外的邏輯，而不影響其他路由。
@app.middleware("request", route="/user/<user_id>")
async def check_user_auth(request, user_id):
# 假設我們通過 header 進行身份驗證
if request.headers.get("Authorization") != "Bearer my_token":
return json({"error": "Unauthorized"}, status=401)
print(f"Authenticated request for user {user_id}")
在這個示例中，check_user_auth 中間件只會在訪問 /user/<user_id> 路由時執(zhí)行，檢查請求頭中的 Authorization 字段是否包含有效的 Token。
3. 異常處理中間件
Sanic 中的異常處理中間件允許你捕獲應用程序中的未處理異常，并為用戶提供定制化的錯誤信息。
你可以使用 @app.exception 裝飾器捕獲特定類型的異常，也可以使用中間件全局處理錯誤。
捕獲特定異常
from sanic.exceptions import SanicException
@app.exception(SanicException)
async def handle_sanic_exception(request, exception):
return json({"error": f"Sanic error: {exception}"}, status=500)
捕獲所有異常
你也可以定義一個捕獲所有異常的全局中間件，并返回自定義的錯誤信息。
@app.middleware("response")
async def handle_all_exceptions(request, response):
if hasattr(request, 'exception') and request.exception:
return json({"error": f"Unhandled error: {request.exception}"}, status=500)
return response
在這個例子中，如果某個請求拋出了異常，并且沒有被其他中間件捕獲，handle_all_exceptions 會捕獲到并返回一個自定義的錯誤響應。
4. 異步中間件
由于 Sanic 是基于 asyncio 的異步框架，你可以編寫異步的中間件，這樣它們不會阻塞事件循環(huán)。中間件是異步的，這意味著你可以執(zhí)行 I/O 操作（例如數(shù)據(jù)庫查詢或 HTTP 請求）而不會阻塞其他請求的處理。
@app.middleware("request")
async def async_middleware(request):
# 異步操作，例如異步查詢數(shù)據(jù)庫
await asyncio.sleep(1)
print("Async operation completed")
5. 優(yōu)先級和中間件順序
多個中間件會按照定義的順序依次執(zhí)行。在 Sanic 中，中間件的優(yōu)先級是固定的，按照以下順序執(zhí)行：
- 請求中間件：按定義的順序執(zhí)行，最先定義的請求中間件最先執(zhí)行。
- 響應中間件：按定義的順序執(zhí)行，最先定義的響應中間件最先執(zhí)行。
6. 中間件的運行機制
請求階段
請求階段中間件是指那些在請求路由匹配之前執(zhí)行的中間件，通常用于請求預處理，例如請求數(shù)據(jù)解析、驗證、認證等。
響應階段
響應階段的中間件是在響應發(fā)送回客戶端之前執(zhí)行的，通常用于對響應進行修改，例如添加額外的響應頭、修改響應數(shù)據(jù)等。
7. 使用中間件進行身份驗證
在 Web 開發(fā)中，身份驗證是非常常見的需求?？梢允褂弥虚g件來實現(xiàn)身份驗證邏輯。比如，檢查用戶請求頭中的 Token 或 Cookies 是否有效。
@app.middleware("request")
async def check_auth(request):
token = request.headers.get("Authorization")
if token != "Bearer valid_token":
return json({"error": "Unauthorized"}, status=401)
8. 使用中間件實現(xiàn) CORS
跨域資源共享（CORS）是一種允許服務器控制哪些域可以訪問其資源的機制。你可以使用中間件來添加 CORS 頭部。
@app.middleware("response")
async def add_cors_headers(request, response):
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
return response
9. 中間件的多個裝飾器
Sanic 支持多個中間件裝飾器應用在同一個請求或響應階段上。例如，你可以同時應用多個請求中間件。
@app.middleware("request")
async def middleware_one(request):
print("Middleware one executed")
@app.middleware("request")
async def middleware_two(request):
print("Middleware two executed")
在這個例子中，middleware_one 會先執(zhí)行，然后是 middleware_two。
總結
Sanic 中的中間件是非常強大和靈活的，它們使得你能夠在 Web 應用的請求處理流程中插入自定義邏輯。中間件的使用場景非常廣泛，包括請求和響應處理、身份驗證、權限控制、日志記錄、CORS 處理、錯誤處理等。
主要特點：
- 請求中間件：在請求路由處理之前執(zhí)行。
- 響應中間件：在響應返回給客戶端之前執(zhí)行。
- 全局中間件：適用于整個應用。
- 路由中間件：只適用于特定的路由。
- 異步操作：支持異步中間件，可以執(zhí)行 I/O 操作而不會阻塞事件循環(huán)。
通過合理使用中間件，你可以在 Sanic 應用中靈活地處理各種需求，提高代碼的可維護性和擴展性。
統(tǒng)計每個請求耗時
要統(tǒng)計一個請求的執(zhí)行時間，你可以利用 Sanic 中間件來實現(xiàn)。在請求處理的生命周期中，我們可以在請求開始時記錄時間戳，在請求結束時計算并輸出請求的執(zhí)行時間。
1. 使用中間件統(tǒng)計請求執(zhí)行時間
我們可以使用 請求中間件 來記錄每個請求開始時的時間戳，然后使用 響應中間件 來計算并輸出請求的執(zhí)行時間。
示例代碼
from sanic import Sanic
from sanic.response import json
import time
app = Sanic("TimingApp")
# 請求中間件：記錄請求開始的時間
@app.middleware("request")
async def record_start_time(request):
request.ctx.start_time = time.time() # 將開始時間存儲到請求上下文中
# 響應中間件：計算并輸出請求執(zhí)行時間
@app.middleware("response")
async def calculate_execution_time(request, response):
if hasattr(request.ctx, 'start_time'):
end_time = time.time()
execution_time = end_time - request.ctx.start_time # 計算執(zhí)行時間
print(f"Request to {request.url} took {execution_time:.4f} seconds")
return response
# 示例路由
@app.route("/")
async def hello(request):
return json({"message": "Hello, Sanic!"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
2. 代碼解析
record_start_time請求中間件：- 在請求到達應用時觸發(fā)，使用
time.time()獲取當前時間并將其存儲在request.ctx.start_time中。ctx是Sanic提供的上下文對象，用來在請求的生命周期內存儲數(shù)據(jù)。
- 在請求到達應用時觸發(fā)，使用
calculate_execution_time響應中間件：- 在請求完成后觸發(fā)，計算請求執(zhí)行的時間。通過
time.time()獲取當前時間，并與request.ctx.start_time進行比較，得到請求的執(zhí)行時間。 - 輸出請求的執(zhí)行時間，單位是秒。
- 在請求完成后觸發(fā)，計算請求執(zhí)行的時間。通過
示例路由：
- 在
/路由中，我們返回一個簡單的 JSON 響應，測試請求執(zhí)行時間的統(tǒng)計。
- 在
3. 測試
當你運行上面的代碼并訪問 http://localhost:8000/ 時，終端輸出類似以下內容：
Request to / took 0.0023 seconds
這表示請求到 / 的執(zhí)行時間是 0.0023 秒。
4. 更復雜的統(tǒng)計方法
如果你需要更精細的控制（例如，統(tǒng)計多個請求的平均執(zhí)行時間、最慢的請求、最短的請求等），你可以進一步擴展這個功能，記錄統(tǒng)計數(shù)據(jù)到日志文件或者數(shù)據(jù)庫中。
例如，你可以將執(zhí)行時間記錄到一個全局的統(tǒng)計數(shù)據(jù)中：
import time
from sanic import Sanic
from sanic.response import json
from collections import defaultdict
app = Sanic("TimingApp")
request_stats = defaultdict(list) # 用于存儲每個路由的執(zhí)行時間
# 請求中間件：記錄請求開始的時間
@app.middleware("request")
async def record_start_time(request):
request.ctx.start_time = time.time()
# 響應中間件：計算請求執(zhí)行時間并保存
@app.middleware("response")
async def calculate_execution_time(request, response):
if hasattr(request.ctx, 'start_time'):
end_time = time.time()
execution_time = end_time - request.ctx.start_time
print(f"Request to {request.url} took {execution_time:.4f} seconds")
request_stats[request.url].append(execution_time) # 記錄執(zhí)行時間
return response
# 獲取平均執(zhí)行時間的路由
@app.route("/stats")
async def stats(request):
stats = {url: sum(times) / len(times) for url, times in request_stats.items()}
return json(stats)
# 示例路由
@app.route("/")
async def hello(request):
return json({"message": "Hello, Sanic!"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
5. 擴展功能的解釋
- 記錄請求執(zhí)行時間：我們使用一個
defaultdict(list)來記錄每個 URL 請求的執(zhí)行時間。在每次請求完成后，我們將執(zhí)行時間添加到對應 URL 的列表中。 - 統(tǒng)計平均執(zhí)行時間：我們創(chuàng)建了一個
/stats路由，用于返回每個路由的平均執(zhí)行時間。
6. 總結
- 使用
Sanic中的 中間件 來統(tǒng)計請求的執(zhí)行時間是非常簡便的。通過記錄請求開始和結束的時間，可以輕松計算出請求的執(zhí)行時間。 - 你可以根據(jù)自己的需求擴展這個功能，如統(tǒng)計最慢的請求、記錄日志等。
這種方法能夠幫助你監(jiān)控應用的性能，尤其是在高并發(fā)的情況下，快速識別出潛在的性能瓶頸。
到此這篇關于Python3中Sanic中間件的使用的文章就介紹到這了,更多相關Python3 Sanic中間件內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了詳解用Python調用百度地圖正/逆地理編碼API，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-07-07
這篇文章主要介紹了python語言time庫和datetime庫基本使用詳解，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-12-12
pyinstaller打包opencv和numpy程序運行錯誤解決
這篇文章主要介紹了pyinstaller打包opencv和numpy程序運行錯誤解決,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-08-08
這篇文章主要介紹了在Python中使用base64模塊處理字符編碼的教程,示例代碼基于Python2.x版本,需要的朋友可以參考下2015-04-04
深入探討PythonLogging模塊的高級用法與性能優(yōu)化
在Python應用程序中，日志處理是一項至關重要的任務，本文將探索Logging模塊的高級用法，包括日志級別、格式化、處理程序等方面的功能，需要的可以參考下2024-04-04
本篇文章主要介紹python 性能提升的幾種方法,并附有代碼參考示例,有需要的小伙伴可以參考下2016-07-07
在mac下查找python包存放路徑site-packages的實現(xiàn)方法
今天小編就為大家分享一篇在mac下查找python包存放路徑site-packages的實現(xiàn)方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-11-11
這篇文章主要給大家介紹了關于OpenCV圖像變換之傅里葉變換的相關資料,傅里葉變換可以將一幅圖片分解為正弦和余弦兩個分量,換而言之,他可以將一幅圖像從其空間域（spatial domain）轉換為頻域（frequency domain）,需要的朋友可以參考下2021-07-07
基于Python實現(xiàn)的影視數(shù)據(jù)智能分析系統(tǒng)
數(shù)據(jù)分析與可視化是當今數(shù)據(jù)分析的發(fā)展方向,大數(shù)據(jù)時代,數(shù)據(jù)資源具有海量特征,數(shù)據(jù)分析和可視化主要通過Python數(shù)據(jù)分析來實現(xiàn),本文給大家介紹了如何基于Python實現(xiàn)的影視數(shù)據(jù)智能分析系統(tǒng),文中給出了部分詳細代碼,感興趣的朋友跟著小編一起來看看吧2024-01-01
python 對給定可迭代集合統(tǒng)計出現(xiàn)頻率,并排序的方法
今天小編就為大家分享一篇python 對給定可迭代集合統(tǒng)計出現(xiàn)頻率,并排序的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-10-10