FastAPI异步安全依赖项

安装

pip install fastapi-async-safe-dependencies

简介

FastAPI是一个构建API的优秀框架，这个库的主要目的是让它变得更好。我猜（像我一样）有人（或团队）在项目中有很多基于类的依赖项。但是FastAPI有一个小问题，这会让你的应用程序比可能的速度慢。

让我们看看FastAPI文档中的一个示例

from fastapi import Depends, FastAPI

app = FastAPI()

fake_items_db = [
    {"item_name": "Foo"},
    {"item_name": "Bar"},
    {"item_name": "Baz"},
]

class CommonQueryParams:
    def __init__(
        self,
        q: str | None = None,
        skip: int = 0,
        limit: int = 100,
    ) -> None:
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip: commons.skip + commons.limit]
    response.update({"items": items})
    return response

这个例子非常简单，并且运行良好，但它有一个小问题。每当FastAPI尝试解析CommonQueryParams依赖项时，它都会将对象创建委托给线程池执行器，这对性能不利。这会在每个请求中发生。

FastAPI的逻辑很简单，如果asyncio.iscoroutinefunction对于依赖项返回True，它将被调用为一个简单的协程，否则，它将被委托给线程池执行器。

FastAPI使用anyio.to_thread.run_sync函数将对象创建委托给线程池执行器。anyio线程池执行器有有限数量的线程（默认为40个），这意味着如果你同时有超过40个请求，其中一些可能会被阻塞，直到池中的一个线程被释放。线程将被简单的类实例化阻塞，这不是CPU密集型操作，这不是你想委托给线程池执行器的任务类型。

本库提供了一个简单的解决方案来解决这个问题，可以避免为基于类的依赖项使用不必要的线程池执行器，并且应该可以提高您的应用程序性能。

使用方法

让我们看看同一个例子，但使用 fastapi-async-safe-dependencies 库

from fastapi import Depends, FastAPI
from fastapi_async_safe import async_safe, init_app

app = FastAPI()
init_app(app)   # don't forget to initialize application

fake_items_db = [
    {"item_name": "Foo"},
    {"item_name": "Bar"},
    {"item_name": "Baz"},
]

@async_safe  # you just need to add this decorator to your dependency
class CommonQueryParams:
    def __init__(
        self,
        q: str | None = None,
        skip: int = 0,
        limit: int = 100,
    ) -> None:
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip: commons.skip + commons.limit]
    response.update({"items": items})
    return response

就是这样，现在您的依赖项将被调用为一个简单的协程，并且不会被委托到线程池。

上面的代码等同于以下代码

from fastapi import Depends, FastAPI

app = FastAPI()

fake_items_db = [
    {"item_name": "Foo"},
    {"item_name": "Bar"},
    {"item_name": "Baz"},
]

class CommonQueryParams:
    def __init__(
        self,
        q: str | None = None,
        skip: int = 0,
        limit: int = 100,
    ) -> None:
        self.q = q
        self.skip = skip
        self.limit = limit


async def _common_query_params(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
) -> CommonQueryParams:
    return CommonQueryParams(q=q, skip=skip, limit=limit)


@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(_common_query_params)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip: commons.skip + commons.limit]
    response.update({"items": items})
    return response

文档

您只需要将 @async_safe 装饰器添加到您的依赖类或不应委托到线程池执行器的同步函数中。

from typing import Any
from dataclasses import dataclass

from fastapi_async_safe import async_safe

@dataclass
@async_safe
class CommonQueryParams:
    q: str | None = None
    skip: int = 0
    limit: int = 100


@async_safe
def common_query_params(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
) -> dict[str, Any]:
    return {"q": q, "skip": skip, "limit": limit}

CommonQueryParams 类和 common_query_params 函数都不会被委托到线程池执行器。

如果您的类继承自被 @async_safe 装饰器装饰的类，则该类也将是 async-safe。

from dataclasses import dataclass

from fastapi_async_safe import async_safe

@dataclass
@async_safe
class BaseQueryParams:
    q: str | None = None


@dataclass  # this class will be async-safe too
class CommonQueryParams(BaseQueryParams):
    skip: int = 0
    limit: int = 100

如果您不希望继承的类是 async-safe，可以使用 @async_unsafe 装饰器。

from dataclasses import dataclass

from fastapi_async_safe import async_safe, async_unsafe

@dataclass
@async_safe
class BaseQueryParams:
    q: str | None = None


@dataclass
@async_unsafe  # this class no longer will be async-safe
class CommonQueryParams(BaseQueryParams):
    skip: int = 0
    limit: int = 100

此外，别忘了使用 init_app 函数初始化您的应用程序，否则 @async_safe 装饰器将没有任何效果。init_app 函数将在应用程序启动时对 Dependant 实例进行猴子补丁。

from fastapi import FastAPI
from fastapi_async_safe import init_app

app = FastAPI()
init_app(app)  # don't forget to initialize application !!!

如果您想将所有基于类的依赖项都使用 @async_safe 装饰器包装，可以将 all_classes_safe=True 参数传递给 init_app 函数。它将包装您所有的基于类的依赖项，除了那些被 @async_unsafe 装饰器装饰的依赖项。

from fastapi import FastAPI
from fastapi_async_safe import init_app

app = FastAPI()
init_app(app, all_classes_safe=True)

基准测试

请查看 基准测试目录 以获取更多详细信息。

性能提升取决于您的应用程序中基于类的依赖项数量，依赖项越多，性能提升越大。