python-3.x 如何在类中使用FastAPI创建路由

2vuwiymt  于 2023-02-10  发布在  Python
关注(0)|答案(8)|浏览(430)

因此,我需要在类中有一些路由,但是路由方法需要有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,而无需将其作为必需参数。

yfjy0ee7

yfjy0ee71#

这可以通过使用APIRouteradd_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的外部依赖性。
1.从调用者发送的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"}
pobjuy32

pobjuy322#

为了创建基于类的视图,你可以使用fastapi-utils中的@cbv decorator。
停止在相关端点的签名中一遍又一遍地重复相同的依赖关系。
您的示例可以重写如下:

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)
ikfrs5lh

ikfrs5lh3#

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

$ 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}

注意,你不需要在每个方法周围都 Package 装饰器,根据它们在HTTP协议中的用途来命名方法就足够了,整个类都变成了一个装饰器。

mcvgt66p

mcvgt66p4#

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

ztigrdn85#

我刚刚发布了一个项目,它允许你使用一个类示例来进行简单装饰器的路由处理。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)

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

jdzmm42g

jdzmm42g6#

在这种情况下,我可以使用python类连接控制器,并使用一个协作者通过dep注入传递它。
Here full example plus tests

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)
gv8xihay

gv8xihay7#

另一种方法是使用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",
    )
)
nkhmeac6

nkhmeac68#

在类中继承FastAPI,并使用FastAPI装饰器作为方法调用(我将使用APIRouter演示它,但您的示例应该可以正常工作):

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

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

相关问题