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

Question

所以我需要在 class 中包含一些路由，但是路由方法需要具有 self 属性（以访问 class' 属性）。但是，FastAPI 然后假设 self 是它自己的必需参数并将其作为查询参数放入

这是我得到的：

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

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

不过，这个returns422除非你去/somewhere?self=something。这个问题是 self 然后是 str，因此没用。

我需要一些方法来访问 self 而无需将其作为必需参数。

Answer 1

要创建 class-based 视图，您可以使用 @cbv decorator from fastapi-utils。使用它的动机：

Stop repeating the same dependencies over and over in the signature of related endpoints.

您的示例可以这样重写：

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)

Answer 2

您在 class 中继承自 FastAPI 并将 FastAPI 装饰器用作方法调用（我将使用 APIRouter 来展示它，但您的示例应该可以正常工作）：

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

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

Answer 3

我把路由放到 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)

Answer 4

我不喜欢这样做的标准方式，所以我编写了自己的库。你可以这样安装它：

$ 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 协议中的用途来命名这些方法就足够了。整个class变成装饰器

Answer 5

我刚刚发布了一个项目，您可以使用 class 实例通过简单的装饰器进行路由处理。 cbv 很酷，但路由是在 class 本身上，而不是 class 的实例。能够使用 class 实例可以让你以一种对我来说更简单、更直观的方式进行依赖注入。

例如，以下内容按预期工作：

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)

您可以 find it on PyPi 并通过 pip install classy-fastapi 安装。

Answer 6

另一种方法是 decorator class that takes parameters。路线是在运行之前注册并添加的：

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",
    )
)

Answer 7

这可以通过使用 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 文档中也称为 path operation function（以下称为“POF”）。

为什么装饰方法不起作用

用 @app.get 和 class 主体中的朋友修饰方法不起作用，因为您会有效地传递 Hello.hello，而不是 hello.hello（a.k.a。self.hello) 到 add_api_route。绑定和未绑定方法（a.k.a 简称为“函数”since Python 3）具有不同的签名：

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

FastAPI 做了很多魔术来尝试自动将 HTTP 请求中的数据（正文或查询参数）解析为 POF 实际使用的对象。

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

假设包含路由的 class 的性质，并即时生成 self（a.k.a 调用 Hello.__init__）。这可能会给 FastAPI 增加很多复杂性，并且是 FastAPI 开发人员（可以理解）似乎没有兴趣支持的用例。似乎处理 application/resource 状态的推荐方法是将整个问题推迟到外部依赖 Depends.
能够从调用者发送的 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"}

Answer 8

在这种情况下，我可以使用 python class 连接控制器，并使用合作者通过 dep 注入传递它。

[这里是完整的例子加上测试]https://github.com/sabatinim/fast_api_hello_world

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)

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

How to create routes with FastAPI within a class

python

python-3.x

class

self

fastapi

为什么装饰方法不起作用

从查询参数