FastAPI高級(jí)教程（一）
共 8262字，需瀏覽 17分鐘
·
2024-08-05 21:23
本文將會(huì)介紹FastAPI中的高級(jí)使用技巧，包括中間件、后臺(tái)任務(wù)、請(qǐng)求前/后、生命周期等。
歡迎關(guān)注我的公眾號(hào)NLP奇幻之旅，原創(chuàng)技術(shù)文章第一時(shí)間推送。
歡迎關(guān)注我的知識(shí)星球“自然語(yǔ)言處理奇幻之旅”，筆者正在努力構(gòu)建自己的技術(shù)社區(qū)。
本文將會(huì)介紹FastAPI中的高級(jí)使用技巧，如下：
中間件（middleware）
后臺(tái)任務(wù)(BackgroundTask)
請(qǐng)求前/后(before or after request)
生命周期(lifespan)
這些技巧將會(huì)使我們更加充分地利用FastAPI的性能，同時(shí)提升服務(wù)的穩(wěn)定性，使項(xiàng)目更健壯、優(yōu)雅，代碼結(jié)構(gòu)更加清晰。
中間件(middleware)
在FastAPI中，中間件是用于在處理請(qǐng)求和生成響應(yīng)之間執(zhí)行代碼的組件。中間件可以用于各種任務(wù)，例如日志記錄、請(qǐng)求和響應(yīng)的修改、身份驗(yàn)證、跨域資源共享 (CORS) 等。
中間件介紹
"中間件"是一個(gè)函數(shù),它在每個(gè)請(qǐng)求被特定的路徑操作處理之前,以及在每個(gè)響應(yīng)返回之前工作。
它接收你的應(yīng)用程序的每一個(gè)
請(qǐng)求.然后它可以對(duì)這個(gè)
請(qǐng)求做一些事情或者執(zhí)行任何需要的代碼.然后它將
請(qǐng)求傳遞給應(yīng)用程序的其他部分 (通過(guò)某種路徑操作).然后它獲取應(yīng)用程序生產(chǎn)的
響應(yīng)(通過(guò)某種路徑操作).它可以對(duì)該
響應(yīng)做些什么或者執(zhí)行任何需要的代碼.然后它返回這個(gè)
響應(yīng).
創(chuàng)建中間件
要?jiǎng)?chuàng)建中間件你可以在函數(shù)的頂部使用裝飾器 @app.middleware("http").
中間件參數(shù)接收如下參數(shù):
request一個(gè)函數(shù)
call_next, 它將接收 request 作為參數(shù).
這個(gè)函數(shù)將
request傳遞給相應(yīng)的 路徑操作.然后它將返回由相應(yīng)的路徑操作生成的
response.
然后你可以在返回
response前進(jìn)一步修改它.
# -*- coding: utf-8 -*-
import time
import random
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
@app.get("/")
async def read_root():
time.sleep(random.random())
return {"message": "Hello, World!"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=18000)
注: 用'X-' 前綴添加專(zhuān)有自定義請(qǐng)求頭。
其它實(shí)現(xiàn)方法
也可以用下面的Python代碼實(shí)現(xiàn)中間件：
# -*- coding: utf-8 -*-
import time
import random
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
print(f"Request processed in {process_time} seconds")
return response
app.add_middleware(TimingMiddleware)
@app.get("/")
async def read_root():
time.sleep(random.random())
return {"message": "Hello, World!"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=18000)
后臺(tái)任務(wù)(BackgroundTask)
你可以定義在返回響應(yīng)后運(yùn)行的后臺(tái)任務(wù)。這對(duì)需要在請(qǐng)求之后執(zhí)行的操作很有用，但客戶(hù)端不必在接收響應(yīng)之前等待操作完成。下面是一些例子：
執(zhí)行操作后發(fā)送的電子郵件通知：
由于連接到電子郵件服務(wù)器并發(fā)送電子郵件往往很“慢”（幾秒鐘），您可以立即返回響應(yīng)并在后臺(tái)發(fā)送電子郵件通知。
處理數(shù)據(jù)：
例如，假設(shè)您收到的文件必須經(jīng)過(guò)一個(gè)緩慢的過(guò)程，您可以返回一個(gè)"Accepted"(HTTP 202)響應(yīng)并在后臺(tái)處理它。
示例Python實(shí)現(xiàn)函數(shù)：
# -*- coding: utf-8 -*-
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=18000)
注: 如果您需要執(zhí)行繁重的后臺(tái)計(jì)算，并且不一定需要由同一進(jìn)程運(yùn)行（例如，您不需要共享內(nèi)存、變量等），那么使用其他更大的工具（如 Celery）可能更好。
請(qǐng)求前/后(before or after request)
在 FastAPI 中，雖然沒(méi)有直接的 before_request 鉤子（如 Flask 中的 before_request），但你可以使用中間件（middleware）來(lái)實(shí)現(xiàn)類(lèi)似的功能。中間件允許你在每個(gè)請(qǐng)求處理之前和之后執(zhí)行代碼。
以下是一個(gè)簡(jiǎn)單的示例，演示如何在 FastAPI 中使用中間件來(lái)實(shí)現(xiàn)類(lèi)似 before_request 的功能。
# -*- coding: utf-8 -*-
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class TraceIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
trace_id = request.headers.get('trace_id')
if trace_id:
print(f"Received trace_id: {trace_id}")
else:
print("No trace_id found in headers")
response = await call_next(request)
return response
app.add_middleware(TraceIDMiddleware)
@app.get("/items/")
async def read_items():
return {"message": "Hello, World!"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=18000)
請(qǐng)求1:
curl http://localhost:18000/items/
在IDE中的輸出結(jié)果為：
No trace_id found in headers
INFO: 127.0.0.1:51063 - "GET /items/ HTTP/1.1" 200 OK
請(qǐng)求2:
curl http://localhost:18000/items/ --header 'trace_id: 123'
在IDE中的輸出結(jié)果為：
Received trace_id: 123
INFO: 127.0.0.1:51108 - "GET /items/ HTTP/1.1" 200 OK
生命周期(lifespan)
您可以定義在應(yīng)用程序 啟動(dòng) 之前應(yīng)該執(zhí)行的邏輯（代碼）。這意味著此代碼將 執(zhí)行一次，在應(yīng)用程序 開(kāi)始接收請(qǐng)求 之前。
同樣，您也可以定義在應(yīng)用程序 關(guān)閉 時(shí)應(yīng)該執(zhí)行的邏輯（代碼）。在這種情況下，此代碼將 執(zhí)行一次，在可能處理了 許多請(qǐng)求 之后。
因?yàn)榇舜a是在應(yīng)用程序 開(kāi)始 接收請(qǐng)求之前執(zhí)行的，并且在它 完成 處理請(qǐng)求之后立即執(zhí)行，所以它涵蓋了整個(gè)應(yīng)用程序 生命周期。
這對(duì)于設(shè)置您需要在整個(gè)應(yīng)用程序中使用的 資源 非常有用，這些資源在請(qǐng)求之間是 共享 的，或者您需要在之后 清理 它們。例如，數(shù)據(jù)庫(kù)連接池，或加載共享的機(jī)器學(xué)習(xí)模型。
示例Python實(shí)現(xiàn)代碼：
# -*- coding: utf-8 -*-
from contextlib import asynccontextmanager
from fastapi import FastAPI
def fake_answer_to_everything_ml_model(x: float):
return x * 42
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Load the ML model
ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
yield
# Clean up the ML models and release the resources
ml_models.clear()
app = FastAPI(lifespan=lifespan)
@app.get("/predict")
async def predict(x: float):
result = ml_models["answer_to_everything"](x)
return {"result": result}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=18000)
請(qǐng)求1:
curl http://localhost:18000/predict\?x\=2
輸出:
{"result":84.0}