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

问题描述 投票：0回答：9

所以我需要在类内有一些路由，但路由方法需要具有 

self
attr （以访问类的属性）。 然而，FastAPI 然后假设 
self
是它自己的必需参数，并将其作为查询参数放入

这就是我所拥有的：

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

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

但是，除非您转到 

422
，否则这会返回 
/somewhere?self=something
。问题是，
self
是 str，因此毫无用处。

我需要某种方法，让我仍然可以访问 

self
，而无需将其作为必需参数。

python python-3.x class self fastapi
9个回答
49
投票

这可以通过使用 

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 起）具有不同的签名：

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"}
17
投票

要创建基于类的视图，您可以使用 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)
9
投票

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

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

7
投票

我刚刚发布了一个项目，允许您使用类 instance 来通过简单的装饰器进行路由处理。 

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
安装。

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)
1
投票

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

这里有完整的示例和测试

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)
0
投票

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

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",
    )
)
0
投票

仅供记录，上面的最佳答案（@Gustavo Perena）也可以实现为：

from fastapi import FastAPI, APIRouter


class Hello:

    def __init__(self, name: str):
        self.name = name
        self.router = APIRouter()
        self.router.get("/hello")(self.hello) # use decorator

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


app = FastAPI()
hello = Hello("World")
app.include_router(hello.router)
-3
投票

您在类中继承了 FastAPI，并使用 FastAPI 装饰器作为方法调用（我将使用 

APIRouter
来展示它，但您的示例应该可以正常工作）：

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

        self.include_router(
            health.router,
            prefix="/api/v1/health",
        )
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.