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 作为参数.

  1. 这个函数将request传递给相应的 路径操作.

  2. 然后它将返回由相应的路径操作生成的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-' 前缀添加专有自定义请求头。

响应体的头部中新增x-process-time参数

其它实现方法

也可以用下面的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}


浏览 75
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报