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

Sanic 中間件的工作流程

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

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

中間件的使用

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

1. 全局中間件

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

請求階段的全局中間件

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

# 響應(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)

在這個(gè)例子中：

• log_request 中間件會(huì)在每個(gè)請求到達(dá)應(yīng)用時(shí)被調(diào)用，打印請求的方法和 URL。
• log_response 中間件會(huì)在每個(gè)響應(yīng)返回給客戶端之前被調(diào)用，打印響應(yīng)的狀態(tài)碼，并在響應(yīng)頭中添加一個(gè)自定義的 header。

2. 路由中間件

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

@app.middleware("request", route="/user/<user_id>")
async def check_user_auth(request, user_id):
    # 假設(shè)我們通過 header 進(jìn)行身份驗(yàn)證
    if request.headers.get("Authorization") != "Bearer my_token":
        return json({"error": "Unauthorized"}, status=401)
    print(f"Authenticated request for user {user_id}")

在這個(gè)示例中，check_user_auth 中間件只會(huì)在訪問 /user/<user_id> 路由時(shí)執(zhí)行，檢查請求頭中的 Authorization 字段是否包含有效的 Token。

3. 異常處理中間件

Sanic 中的異常處理中間件允許你捕獲應(yīng)用程序中的未處理異常，并為用戶提供定制化的錯(cuò)誤信息。

你可以使用 @app.exception 裝飾器捕獲特定類型的異常，也可以使用中間件全局處理錯(cuò)誤。

捕獲特定異常

from sanic.exceptions import SanicException

@app.exception(SanicException)
async def handle_sanic_exception(request, exception):
    return json({"error": f"Sanic error: {exception}"}, status=500)

捕獲所有異常

你也可以定義一個(gè)捕獲所有異常的全局中間件，并返回自定義的錯(cuò)誤信息。

@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

在這個(gè)例子中，如果某個(gè)請求拋出了異常，并且沒有被其他中間件捕獲，handle_all_exceptions 會(huì)捕獲到并返回一個(gè)自定義的錯(cuò)誤響應(yīng)。

4. 異步中間件

由于 Sanic 是基于 asyncio 的異步框架，你可以編寫異步的中間件，這樣它們不會(huì)阻塞事件循環(huán)。中間件是異步的，這意味著你可以執(zhí)行 I/O 操作（例如數(shù)據(jù)庫查詢或 HTTP 請求）而不會(huì)阻塞其他請求的處理。

@app.middleware("request")
async def async_middleware(request):
    # 異步操作，例如異步查詢數(shù)據(jù)庫
    await asyncio.sleep(1)
    print("Async operation completed")

5. 優(yōu)先級和中間件順序

多個(gè)中間件會(huì)按照定義的順序依次執(zhí)行。在 Sanic 中，中間件的優(yōu)先級是固定的，按照以下順序執(zhí)行：

• 請求中間件：按定義的順序執(zhí)行，最先定義的請求中間件最先執(zhí)行。
• 響應(yīng)中間件：按定義的順序執(zhí)行，最先定義的響應(yīng)中間件最先執(zhí)行。

6. 中間件的運(yùn)行機(jī)制

請求階段

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

響應(yīng)階段

響應(yīng)階段的中間件是在響應(yīng)發(fā)送回客戶端之前執(zhí)行的，通常用于對響應(yīng)進(jìn)行修改，例如添加額外的響應(yīng)頭、修改響應(yīng)數(shù)據(jù)等。

7. 使用中間件進(jìn)行身份驗(yàn)證

在 Web 開發(fā)中，身份驗(yàn)證是非常常見的需求?？梢允褂弥虚g件來實(shí)現(xiàn)身份驗(yà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. 使用中間件實(shí)現(xiàn) CORS

跨域資源共享（CORS）是一種允許服務(wù)器控制哪些域可以訪問其資源的機(jī)制。你可以使用中間件來添加 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. 中間件的多個(gè)裝飾器

Sanic 支持多個(gè)中間件裝飾器應(yīng)用在同一個(gè)請求或響應(yīng)階段上。例如，你可以同時(shí)應(yīng)用多個(gè)請求中間件。

@app.middleware("request")
async def middleware_one(request):
    print("Middleware one executed")

@app.middleware("request")
async def middleware_two(request):
    print("Middleware two executed")

在這個(gè)例子中，middleware_one 會(huì)先執(zhí)行，然后是 middleware_two。

總結(jié)

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

主要特點(diǎn)：

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

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

統(tǒng)計(jì)每個(gè)請求耗時(shí)

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

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

我們可以使用 請求中間件 來記錄每個(gè)請求開始時(shí)的時(shí)間戳，然后使用 響應(yīng)中間件 來計(jì)算并輸出請求的執(zhí)行時(shí)間。

示例代碼

from sanic import Sanic
from sanic.response import json
import time

app = Sanic("TimingApp")

# 請求中間件：記錄請求開始的時(shí)間
@app.middleware("request")
async def record_start_time(request):
    request.ctx.start_time = time.time()  # 將開始時(shí)間存儲(chǔ)到請求上下文中

# 響應(yīng)中間件：計(jì)算并輸出請求執(zhí)行時(shí)間
@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  # 計(jì)算執(zhí)行時(shí)間
        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 請求中間件：

  • 在請求到達(dá)應(yīng)用時(shí)觸發(fā)，使用 time.time() 獲取當(dāng)前時(shí)間并將其存儲(chǔ)在 request.ctx.start_time 中。ctx 是 Sanic 提供的上下文對象，用來在請求的生命周期內(nèi)存儲(chǔ)數(shù)據(jù)。

• calculate_execution_time 響應(yīng)中間件：

  • 在請求完成后觸發(fā)，計(jì)算請求執(zhí)行的時(shí)間。通過 time.time() 獲取當(dāng)前時(shí)間，并與 request.ctx.start_time 進(jìn)行比較，得到請求的執(zhí)行時(shí)間。
  • 輸出請求的執(zhí)行時(shí)間，單位是秒。

• 示例路由：

  • 在 / 路由中，我們返回一個(gè)簡單的 JSON 響應(yīng)，測試請求執(zhí)行時(shí)間的統(tǒng)計(jì)。

3. 測試

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

Request to / took 0.0023 seconds

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

4. 更復(fù)雜的統(tǒng)計(jì)方法

如果你需要更精細(xì)的控制（例如，統(tǒng)計(jì)多個(gè)請求的平均執(zhí)行時(shí)間、最慢的請求、最短的請求等），你可以進(jìn)一步擴(kuò)展這個(gè)功能，記錄統(tǒng)計(jì)數(shù)據(jù)到日志文件或者數(shù)據(jù)庫中。

例如，你可以將執(zhí)行時(shí)間記錄到一個(gè)全局的統(tǒng)計(jì)數(shù)據(jù)中：

import time
from sanic import Sanic
from sanic.response import json
from collections import defaultdict

app = Sanic("TimingApp")
request_stats = defaultdict(list)  # 用于存儲(chǔ)每個(gè)路由的執(zhí)行時(shí)間

# 請求中間件：記錄請求開始的時(shí)間
@app.middleware("request")
async def record_start_time(request):
    request.ctx.start_time = time.time()

# 響應(yīng)中間件：計(jì)算請求執(zhí)行時(shí)間并保存
@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í)行時(shí)間

    return response

# 獲取平均執(zhí)行時(shí)間的路由
@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. 擴(kuò)展功能的解釋

• 記錄請求執(zhí)行時(shí)間：我們使用一個(gè) defaultdict(list) 來記錄每個(gè) URL 請求的執(zhí)行時(shí)間。在每次請求完成后，我們將執(zhí)行時(shí)間添加到對應(yīng) URL 的列表中。
• 統(tǒng)計(jì)平均執(zhí)行時(shí)間：我們創(chuàng)建了一個(gè) /stats 路由，用于返回每個(gè)路由的平均執(zhí)行時(shí)間。

6. 總結(jié)

• 使用 Sanic 中的 中間件 來統(tǒng)計(jì)請求的執(zhí)行時(shí)間是非常簡便的。通過記錄請求開始和結(jié)束的時(shí)間，可以輕松計(jì)算出請求的執(zhí)行時(shí)間。
• 你可以根據(jù)自己的需求擴(kuò)展這個(gè)功能，如統(tǒng)計(jì)最慢的請求、記錄日志等。

這種方法能夠幫助你監(jiān)控應(yīng)用的性能，尤其是在高并發(fā)的情況下，快速識別出潛在的性能瓶頸。

到此這篇關(guān)于Python3中Sanic中間件的使用的文章就介紹到這了,更多相關(guān)Python3 Sanic中間件內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論