FastAPI高级教程(一)
共 8262字,需浏览 17分钟
·
2024-08-05 21:23
本文将会介绍FastAPI中的高级使用技巧,包括中间件、后台任务、请求前/后、生命周期等。
欢迎关注我的公众号NLP奇幻之旅,原创技术文章第一时间推送。
欢迎关注我的知识星球“自然语言处理奇幻之旅”,笔者正在努力构建自己的技术社区。
本文将会介绍FastAPI
中的高级使用技巧,如下:
中间件(middleware)
后台任务(BackgroundTask)
请求前/后(before or after request)
生命周期(lifespan)
这些技巧将会使我们更加充分地利用FastAPI的性能,同时提升服务的稳定性,使项目更健壮、优雅,代码结构更加清晰。
中间件(middleware)
在FastAPI
中,中间件是用于在处理请求和生成响应之间执行代码的组件。中间件可以用于各种任务,例如日志记录、请求和响应的修改、身份验证、跨域资源共享 (CORS) 等。
中间件介绍
"中间件"是一个函数,它在每个请求
被特定的路径操作处理之前,以及在每个响应
返回之前工作。
它接收你的应用程序的每一个
请求
.然后它可以对这个
请求
做一些事情或者执行任何需要的代码.然后它将
请求
传递给应用程序的其他部分 (通过某种路径操作).然后它获取应用程序生产的
响应
(通过某种路径操作).它可以对该
响应
做些什么或者执行任何需要的代码.然后它返回这个
响应
.
创建中间件
要创建中间件你可以在函数的顶部使用装饰器 @app.middleware("http").
中间件参数接收如下参数:
request
一个函数
call_next
, 它将接收 request 作为参数.
这个函数将
request
传递给相应的 路径操作.然后它将返回由相应的路径操作生成的
response
.
然后你可以在返回
response
前进一步修改它.
# -*- 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-' 前缀添加专有自定义请求头。
其它实现方法
也可以用下面的Python代码实现中间件:
# -*- 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)
后台任务(BackgroundTask)
你可以定义在返回响应后运行的后台任务。这对需要在请求之后执行的操作很有用,但客户端不必在接收响应之前等待操作完成。下面是一些例子:
执行操作后发送的电子邮件通知:
由于连接到电子邮件服务器并发送电子邮件往往很“慢”(几秒钟),您可以立即返回响应并在后台发送电子邮件通知。
处理数据:
例如,假设您收到的文件必须经过一个缓慢的过程,您可以返回一个"Accepted"(HTTP 202)响应并在后台处理它。
示例Python实现函数:
# -*- 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)
注: 如果您需要执行繁重的后台计算,并且不一定需要由同一进程运行(例如,您不需要共享内存、变量等),那么使用其他更大的工具(如 Celery
)可能更好。
请求前/后(before or after request)
在 FastAPI 中,虽然没有直接的 before_request 钩子(如 Flask 中的 before_request),但你可以使用中间件(middleware)来实现类似的功能。中间件允许你在每个请求处理之前和之后执行代码。
以下是一个简单的示例,演示如何在 FastAPI 中使用中间件来实现类似 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)
请求1:
curl http://localhost:18000/items/
在IDE中的输出结果为:
No trace_id found in headers
INFO: 127.0.0.1:51063 - "GET /items/ HTTP/1.1" 200 OK
请求2:
curl http://localhost:18000/items/ --header 'trace_id: 123'
在IDE中的输出结果为:
Received trace_id: 123
INFO: 127.0.0.1:51108 - "GET /items/ HTTP/1.1" 200 OK
生命周期(lifespan)
您可以定义在应用程序 启动 之前应该执行的逻辑(代码)。这意味着此代码将 执行一次,在应用程序 开始接收请求 之前。
同样,您也可以定义在应用程序 关闭 时应该执行的逻辑(代码)。在这种情况下,此代码将 执行一次,在可能处理了 许多请求 之后。
因为此代码是在应用程序 开始 接收请求之前执行的,并且在它 完成 处理请求之后立即执行,所以它涵盖了整个应用程序 生命周期。
这对于设置您需要在整个应用程序中使用的 资源 非常有用,这些资源在请求之间是 共享 的,或者您需要在之后 清理 它们。例如,数据库连接池,或加载共享的机器学习模型。
示例Python实现代码:
# -*- 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)
请求1:
curl http://localhost:18000/predict\?x\=2
输出:
{"result":84.0}