如何在FastAPI中使用类创建路由

Question

所以我需要在一个类的内部设置一些路由，但是路由方法需要有 self属性(以便访问类的属性)。然而，FastAPI会认为self是必须的参数，并将其作为查询参数输入。

这就是我得到的:

app = FastAPI()
class Foo:
    def __init__(y: int):
        self.x = y

    @app.get("/somewhere")
    def bar(self): return self.x

不过，除非您跳转到/somewhere?self=something，否则这将返回422。问题在于self然后是str，因此无用。

我需要一些方法，可以在没有它作为必需参数的情况下仍然访问self。

- Nexy7574

Answer 1

可以通过使用APIRouter的add_api_route方法来实现：

from fastapi import FastAPI, APIRouter


class Hello:

    def __init__(self, name: str):
        self.name = name
        self.router = APIRouter()
        self.router.add_api_route("/hello", self.hello, methods=["GET"])

    def hello(self):
        return {"Hello": self.name}


app = FastAPI()
hello = Hello("World")
app.include_router(hello.router)

例子：

$ curl 127.0.0.1:5000/hello
{"Hello":"World"}

add_api_route的第二个参数(endpoint)的类型为Callable[..., Any]，因此任何可调用对象都可以使用(只要FastAPI能够找到如何解析其参数的HTTP请求数据)。在FastAPI文档中，这个可调用对象也被称为路径操作函数 (以下简称“POF”)。

为什么修饰方法不起作用

警告：如果您对 OP 答案中的代码为什么不起作用不感兴趣，请忽略本答案的其余部分

在类体中使用 @app.get 等装饰器修饰一个方法是行不通的，因为你实际上将传递Hello.hello，而不是hello.hello（即self.hello）给add_api_route。绑定和非绑定方法（Python 3 中也称为“函数”自 Python 3 起）具有不同的签名:

import inspect
inspect.signature(Hello.hello)  # <Signature (self)>
inspect.signature(hello.hello)  # <Signature ()>

FastAPI会尽力自动解析HTTP请求中的数据（主体或查询参数）为POF实际使用的对象，这是通过很多魔法实现的。

如果使用未绑定方法（=常规函数）(Hello.hello)作为POF，FastAPI必须：

1. 对包含路由的类的性质做出假设并即时生成self(别名调用Hello.__init__)。这可能会给FastAPI增加很多复杂性，而且FastAPI开发人员似乎不感兴趣支持这种用例。处理应用程序/资源状态的推荐方式似乎是将整个问题延迟到具有Depends的外部依赖项。

2. 以某种方式能够从调用方发送的HTTP请求数据（通常是JSON）生成一个self对象。对于除字符串或其他内置类型之外的任何内容，这在技术上都是不可行的，因此无法真正使用。

OP的代码执行了#2。FastAPI试图从HTTP请求查询参数中解析Hello.hello的第一个参数（=self，类型为Hello），显然失败并引发RequestValidationError，该错误将作为HTTP 422响应显示给调用方。

从查询参数解析self

为了证明上述#2，以下是一个（无用的）示例，展示了FastAPI实际上可以从HTTP请求中“解析”self：

（免责声明：不要将以下代码用于任何真正的应用程序）

from fastapi import FastAPI

app = FastAPI()

class Hello(str):
    @app.get("/hello")
    def hello(self):
        return {"Hello": self}

例子：

$ curl '127.0.0.1:5000/hello?self=World'
{"Hello":"World"}

- Gustavo Bezerra

---

再次问候。我在 Stack Overflow 上发布了一个有关如何解决依赖注入问题的通用问题，当参数来自基于类的 fastapi 的 CLI 时。我仍在使用您的方法，但需要此功能并不知道如何使其正常工作。该问题位于 https://stackoverflow.com/questions/75449717/fastapi-dependency-injection-with-cli-arguments。 - Henry Thornton

Answer 2

要创建基于类的视图，您可以使用fastapi-utils中的@cbv装饰器。使用它的动机是:

在相关端点的签名中停止重复相同的依赖项。

您的示例可以重写为:

from fastapi import Depends, FastAPI
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter


def get_x():
    return 10


app = FastAPI()
router = InferringRouter()  # Step 1: Create a router


@cbv(router)  # Step 2: Create and decorate a class to hold the endpoints
class Foo:
    # Step 3: Add dependencies as class attributes
    x: int = Depends(get_x)

    @router.get("/somewhere")
    def bar(self) -> int:
        # Step 4: Use `self.<dependency_name>` to access shared dependencies
        return self.x


app.include_router(router)

- alex_noname

---

如果您将“session”作为共享依赖项，那么并发请求会共享同一个实例吗？ - AndreFeijo

创建类实例并独立调用每个请求的依赖项。 - alex_noname

当我尝试注入我的自定义类的实例时，它抛出一个错误，说它应该是一个 Pydantic 感知的类型 o_O 这是否是预期的？ - Stanislav Bashkyrtsev

Answer 3

我不喜欢标准的做法，所以我写了自己的库。你可以像这样安装它:

$ pip install cbfa

以下是如何使用它的示例：

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel
from cbfa import ClassBased


app = FastAPI()
wrapper = ClassBased(app)

class Item(BaseModel):
    name: str
    price: float
    is_offer: Optional[bool] = None

@wrapper('/item')
class Item:
    def get(item_id: int, q: Optional[str] = None):
        return {"item_id": item_id, "q": q}

    def post(item_id: int, item: Item):
        return {"item_name": item.name, "item_id": item_id}

请注意，您不需要在每个方法周围包装装饰器。只需根据它们在HTTP协议中的用途命名方法即可。整个类将被转换为一个装饰器。

- Evgeniy Blinov

---

我非常喜欢这种方法。它支持异步路由吗？ - Austin Hallett

是的，它将适用于所有类型的路由。 - Evgeniy Blinov

Answer 4

我刚发布了一个项目，它允许你使用一个类的实例来处理路由，并且使用简单的装饰器。虽然cbv很棒，但路由是在类本身上定义的，而不是在类的实例上定义的。能够使用类的实例可以让你以一种更简单、更直观的方式进行依赖注入。

例如，下面的代码可以正常工作：

from classy_fastapi import Routable, get, delete

class UserRoutes(Routable):
   """Inherits from Routable."""

   # Note injection here by simply passing values
   # to the constructor. Other injection frameworks also 
   # supported as there's nothing special about this __init__ method.
   def __init__(self, dao: Dao) -> None:
      """Constructor. The Dao is injected here."""
      super().__init__()
      self.__dao = Dao

   @get('/user/{name}')
   def get_user_by_name(name: str) -> User:
      # Use our injected DAO instance.
      return self.__dao.get_user_by_name(name)

   @delete('/user/{name}')
   def delete_user(name: str) -> None:
      self.__dao.delete(name)


def main():
    args = parse_args()
    # Configure the DAO per command line arguments
    dao = Dao(args.url, args.user, args.password)
    # Simple intuitive injection
    user_routes = UserRoutes(dao)
    
    app = FastAPI()
    # router member inherited from Routable and configured per the annotations.
    app.include_router(user_routes.router)

你可以在PyPi上找到它，并通过pip install classy-fastapi来安装。

- Oliver Dain

---

非常感谢您提供的这个软件包！@Olivier，在调用include_router时，我遇到了一个错误，说没有'router属性'。在__init__()中，不应该调用super().init()吗？如果是的话，这个例子也在GitLab自述文件中（这样您就不会忘记了）。 - Sarye Haddadi

@mpc-DT 谢谢你注意到了这个问题。我会修复它！ - Oliver Dain

@OliverDain 在我的情况下，如果__name__ == "main"，则args参数会传递过来，可以调用main()函数或在__main__下进行配置。无论哪种方式，我都会收到“app”找不到的错误。将app=FastAPI()从main()或__main__中提取出来可以解决这个问题，但是会将'app'留作全局变量。我错过了什么吗？ - Henry Thornton

@HenryThornton 很难在没有看到你的代码的情况下做出判断。看起来你漏掉了一些东西，但我不确定是什么。 - Oliver Dain

@OliverDain 在你上面的代码中，app=FastApi()在main()函数中。main()函数是如何被调用的，参数又是如何传递给main()函数的？ - Henry Thornton

Answer 5

我把路由信息放到了def __init__中，它正常工作。示例：

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

class CustomAPI(FastAPI):
    def __init__(self, title: str = "CustomAPI") -> None:
        super().__init__(title=title)

        @self.get('/')
        async def home():
            """
            Home page
            """
            return HTMLResponse("<h1>CustomAPI</h1><br/><a href='/docs'>Try api now!</a>", status_code=status.HTTP_200_OK)

- Khiem Tran

---

这会“中断”IDE中的导航，也就是说，你无法跳转到home()函数，因为它在构造函数的作用域内被声明并且无法访问。 - Tuukka Mustonen

Answer 6

在这种情况下，我可以使用Python类进行控制器的连线，并通过依赖注入传递协作者。 这里是完整的示例和测试。

class UseCase:
    @abstractmethod
    def run(self):
        pass


class ProductionUseCase(UseCase):
    def run(self):
        return "Production Code"


class AppController:

    def __init__(self, app: FastAPI, use_case: UseCase):
        @app.get("/items/{item_id}")
        def read_item(item_id: int, q: Optional[str] = None):
            return {
                "item_id": item_id, "q": q, "use_case": use_case.run()
            }


def startup(use_case: UseCase = ProductionUseCase()):
    app = FastAPI()
    AppController(app, use_case)
    return app


if __name__ == "__main__":
    uvicorn.run(startup(), host="0.0.0.0", port=8080)

- Marco Sabatini

---

生产使用案例类的目的是什么？ - Alejandro

Answer 7

另一种方法是使用带参数的装饰器类。路由在运行时之前注册并添加：

from functools import wraps

_api_routes_registry = []


class api_route(object):
    def __init__(self, path, **kwargs):
        self._path = path
        self._kwargs = kwargs

    def __call__(self, fn):
        cls, method = fn.__repr__().split(" ")[1].split(".")
        _api_routes_registry.append(
            {
                "fn": fn,
                "path": self._path,
                "kwargs": self._kwargs,
                "cls": cls,
                "method": method,
            }
        )

        @wraps(fn)
        def decorated(*args, **kwargs):
            return fn(*args, **kwargs)

        return decorated

    @classmethod
    def add_api_routes(cls, router):
        for reg in _api_routes_registry:
            if router.__class__.__name__ == reg["cls"]:
                router.add_api_route(
                    path=reg["path"],
                    endpoint=getattr(router, reg["method"]),
                    **reg["kwargs"],
                )

定义一个自定义路由器，继承APIRouter并在__init__中添加路由：

class ItemRouter(APIRouter):
    @api_route("/", description="this reads an item")
    def read_item(a: str = "de"):
        return [7262, 324323, a]

    @api_route("/", methods=["POST"], description="add an item")
    def post_item(a: str = "de"):
        return a

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        add_api_routes(self)


app.include_router(
    ItemRouter(
        prefix="/items",
    )
)

- Joost Döbken

Answer 8

你可以在你的类中继承FastAPI，并使用FastAPI装饰器作为方法调用（我将使用APIRouter来展示，但你的例子应该也能工作）：

class Foo(FastAPI):
    def __init__(y: int):
        self.x = y

        self.include_router(
            health.router,
            prefix="/api/v1/health",
        )

- sysid