Python3中Sanic中間件的使用

更新時間：2025年01月08日 10:22:42 作者：言之。

Sanic框架中的中間件是一種強大的工具,本文就來介紹Python3中Sanic中間件的使用,具有一定的參考價值,感興趣的可以了解一下

在 Sanic 中，中間件（middleware）是指在請求和響應(yīng)之間執(zhí)行的代碼。它們是一個非常強大的工具，用于處理請求的預(yù)處理、響應(yīng)的后處理、全局錯誤處理、日志記錄、認證、權(quán)限校驗、跨域資源共享（CORS）等任務(wù)。中間件通常用于應(yīng)用程序的全局邏輯，在請求處理前或響應(yīng)處理后執(zhí)行。

Sanic 中間件的工作流程

Sanic 中間件會在以下兩個階段執(zhí)行：

請求處理前：處理請求數(shù)據(jù)，在請求被路由處理之前執(zhí)行。可以用于校驗、修改請求等操作。
響應(yīng)處理后：處理響應(yīng)數(shù)據(jù)，在響應(yīng)被發(fā)送給客戶端之前執(zhí)行?？梢杂糜谛薷捻憫?yīng)、記錄日志等操作。

中間件的使用

在 Sanic 中，中間件通過裝飾器來定義。你可以為整個應(yīng)用程序（全局中間件）定義中間件，也可以為某些特定的路由定義中間件。

1. 全局中間件

全局中間件是在整個應(yīng)用程序的生命周期中被調(diào)用的，不依賴于特定的路由。它們在請求和響應(yīng)的各個階段都會執(zhí)行。

請求階段的全局中間件

全局中間件通常用于處理請求的預(yù)處理，如身份驗證、記錄日志、設(shè)置 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}")
    # 你可以在這里對請求進行修改，例如身份驗證、設(shè)置 CORS 等
    # 如果請求需要認證，可以在這里進行驗證

# 響應(yīng)階段的全局中間件
@app.middleware("response")
async def log_response(request, response):
    print(f"Response status: {response.status}")
    # 你可以在這里修改響應(yīng)，例如添加 CORS 頭部、記錄日志等
    # 在響應(yīng)返回給客戶端之前處理
    response.headers['X-Custom-Header'] = 'Sanic'  # 可以修改響應(yīng)頭
    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 中間件會在每個請求到達應(yīng)用時被調(diào)用，打印請求的方法和 URL。
log_response 中間件會在每個響應(yīng)返回給客戶端之前被調(diào)用，打印響應(yīng)的狀態(tài)碼，并在響應(yīng)頭中添加一個自定義的 header。

2. 路由中間件

你還可以為特定的路由或路由組定義中間件。這是 Sanic 提供的靈活性，可以讓你為某些路由添加額外的邏輯，而不影響其他路由。

@app.middleware("request", route="/user/<user_id>")
async def check_user_auth(request, user_id):
    # 假設(shè)我們通過 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 中的異常處理中間件允許你捕獲應(yīng)用程序中的未處理異常，并為用戶提供定制化的錯誤信息。

你可以使用 @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 會捕獲到并返回一個自定義的錯誤響應(yīng)。

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í)行。
響應(yīng)中間件：按定義的順序執(zhí)行，最先定義的響應(yīng)中間件最先執(zhí)行。

6. 中間件的運行機制

請求階段

請求階段中間件是指那些在請求路由匹配之前執(zhí)行的中間件，通常用于請求預(yù)處理，例如請求數(shù)據(jù)解析、驗證、認證等。

響應(yīng)階段

響應(yīng)階段的中間件是在響應(yīng)發(fā)送回客戶端之前執(zhí)行的，通常用于對響應(yīng)進行修改，例如添加額外的響應(yīng)頭、修改響應(yīng)數(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）是一種允許服務(wù)器控制哪些域可以訪問其資源的機制。你可以使用中間件來添加 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 支持多個中間件裝飾器應(yīng)用在同一個請求或響應(yīng)階段上。例如，你可以同時應(yīng)用多個請求中間件。

@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。

總結(jié)

Sanic 中的中間件是非常強大和靈活的，它們使得你能夠在 Web 應(yīng)用的請求處理流程中插入自定義邏輯。中間件的使用場景非常廣泛，包括請求和響應(yīng)處理、身份驗證、權(quán)限控制、日志記錄、CORS 處理、錯誤處理等。

主要特點：

請求中間件：在請求路由處理之前執(zhí)行。
響應(yīng)中間件：在響應(yīng)返回給客戶端之前執(zhí)行。
全局中間件：適用于整個應(yīng)用。
路由中間件：只適用于特定的路由。
異步操作：支持異步中間件，可以執(zhí)行 I/O 操作而不會阻塞事件循環(huán)。

通過合理使用中間件，你可以在 Sanic 應(yīng)用中靈活地處理各種需求，提高代碼的可維護性和擴展性。

統(tǒng)計每個請求耗時

要統(tǒng)計一個請求的執(zhí)行時間，你可以利用 Sanic 中間件來實現(xiàn)。在請求處理的生命周期中，我們可以在請求開始時記錄時間戳，在請求結(jié)束時計算并輸出請求的執(zhí)行時間。

1. 使用中間件統(tǒng)計請求執(zhí)行時間

我們可以使用 請求中間件 來記錄每個請求開始時的時間戳，然后使用 響應(yīng)中間件 來計算并輸出請求的執(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()  # 將開始時間存儲到請求上下文中

# 響應(yīng)中間件：計算并輸出請求執(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 請求中間件：
- 在請求到達應(yīng)用時觸發(fā)，使用 time.time() 獲取當(dāng)前時間并將其存儲在 request.ctx.start_time 中。ctx 是 Sanic 提供的上下文對象，用來在請求的生命周期內(nèi)存儲數(shù)據(jù)。
calculate_execution_time 響應(yīng)中間件：
- 在請求完成后觸發(fā)，計算請求執(zhí)行的時間。通過 time.time() 獲取當(dāng)前時間，并與 request.ctx.start_time 進行比較，得到請求的執(zhí)行時間。
- 輸出請求的執(zhí)行時間，單位是秒。
示例路由：
- 在 / 路由中，我們返回一個簡單的 JSON 響應(yīng)，測試請求執(zhí)行時間的統(tǒng)計。

3. 測試

當(dāng)你運行上面的代碼并訪問 http://localhost:8000/ 時，終端輸出類似以下內(nèi)容：

Request to / took 0.0023 seconds

這表示請求到 / 的執(zhí)行時間是 0.0023 秒。

4. 更復(fù)雜的統(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()

# 響應(yīng)中間件：計算請求執(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í)行時間添加到對應(yīng) URL 的列表中。
統(tǒng)計平均執(zhí)行時間：我們創(chuàng)建了一個 /stats 路由，用于返回每個路由的平均執(zhí)行時間。