python 如何在FastAPI中使用基于类的视图?

fcg9iug3  于 2023-01-29  发布在  Python
关注(0)|答案(1)|浏览(312)

我尝试在我的FastApi项目中使用基于类的视图来减少代码冗余。基本上,我需要CRUD功能用于我的所有模型,因此必须一遍又一遍地编写相同的路由。我创建了一个小的示例项目来显示我到目前为止的进展,但我遇到了一些问题。
我知道有这个Fastapi-utils,但据我所知,它只减少了要调用的依赖项的数量,并且不再得到正确的维护(上次提交是在2020年3月)。
我有一些任意的模式/模型。SQLAlchemy模型和DB连接现在是无关紧要的。

from typing import Optional
from pydantic import BaseModel

class ObjBase(BaseModel):
    name: Optional[str]

class ObjCreate(ObjBase):
    pass

class ObjUpdate(ObjBase):
    pass

class Obj(ObjBase):
    id: int

BaseService类用于实现数据库访问,为了简化,现在没有数据库访问,只实现了get(通过id)和list(all)。

from typing import Any, Generic, List, Optional, Type, TypeVar
from pydantic import BaseModel

SchemaType = TypeVar("SchemaType", bound=BaseModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class BaseService(Generic[SchemaType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[SchemaType]):
        self.model = model

    def get(self, id: Any) -> Any:
        return {"id": id}

    def list(self, skip: int = 0, limit: int = 100) -> Any:
        return [
            {"id": 1},
            {"id": 2},
        ]

然后,这个BaseService可以由ObjService类继承,为先前定义的pydantic Obj模型提供这些基本函数。

from schemas.obj import Obj, ObjCreate, ObjUpdate
from .base import BaseService

class ObjService(BaseService[Obj, ObjCreate, ObjUpdate]):
    def __init__(self):
        super(ObjService, self).__init__(Obj)

在此目录的init.py文件中,提供了一个用于获取ObjService示例的函数。

from fastapi import Depends
from .obj import ObjService

def get_obj_service() -> ObjService:
    return ObjService()

到目前为止一切正常。我可以将服务类注入到相关的FastApi路由中。但是所有路由都需要为每个模型和CRUD函数编写。当为多个模型/模式提供相同的API端点时,这会变得很乏味。因此,我的想法是通过提供一个BaseRouter来定义这些路由并为每个模型继承该类,从而使用类似于BaseService背后的逻辑。
基本路由器类:

from typing import Generic, Type, TypeVar
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from services.base import BaseService

SchemaType = TypeVar("SchemaType", bound=BaseModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class BaseRouter(Generic[SchemaType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, schema: Type[SchemaType], prefix: str, service: BaseService):
        self.schema = schema
        self.service = service
        
        self.router = APIRouter(
            prefix=prefix
        )

        self.router.add_api_route("/", self.list, methods=['GET'])
        self.router.add_api_route("/{id}", self.get, methods=['GET'])

    def get(self, id):
        return self.service.get(id)

    def list(self):
        return self.service.list()

ObjRouter类:

from schemas.obj import Obj, ObjCreate, ObjUpdate
from .base import BaseRouter
from services.base import BaseService

class ObjRouter(BaseRouter[Obj, ObjCreate, ObjUpdate]):
    def __init__(self, prefix: str, service: BaseService):
        super(ObjRouter, self).__init__(Obj, prefix, service)

该目录中的init.py文件

from fastapi import Depends
from services import get_obj_service
from services.obj import ObjService
from .obj import ObjRouter

def get_obj_router(service: ObjService = Depends(get_obj_service())) -> ObjRouter:
    return ObjRouter("/obj", service).router

在我的main.py文件中,此路由器被添加到FastApi应用程序中。

from fastapi import Depends, FastAPI
from routes import get_obj_router

app = FastAPI()

app.include_router(get_obj_router())

启动应用程序时,路径Get“/obj”和Get“/obj/id”显示在我的Swagger Docs中。但在测试其中一个端点时,我收到了AttributeError:“Depends”对象没有属性“list”
据我所知,Depends只能用在FastApi函数或本身就是dependies的函数中,因此我尝试修改www.example.com中的app.include_router行main.py,如下所示

app.include_router(Depends(get_obj_router()))

但它再次抛出AttributeError:“Depends”对象没有属性“routes”。
长话短说:我做错了什么?这在FastApi中甚至是可能的吗?或者我需要坚持一遍又一遍地定义相同的CRUD Api端点吗?
我想使用FastApi的Dependenvy注入功能的原因是,稍后我将在Service类中使用以下函数调用来注入DB会话,并在请求后自动关闭它:

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

据我所知,只有当依赖性层次结构中的最高调用(路由依赖于服务依赖于get_db)由FastApi路由完成时,这才是可能的。
PS:这是我在StackOverflow上的第一个问题,请温柔一点。

9rnv2umw

9rnv2umw1#

由于你的问题很长,我会在答案的底部贴出一个完整的例子。
FastAPI中的相关性是可以修改端点参数并将值向下传递给它们的可调用对象。在API模型中,它们在端点级别工作。要传递任何相关性结果,您需要将它们显式传递给控制器函数。
在下面的例子中,我创建了一个虚拟的Session类和一个虚拟的会话注入函数(injecting_session),然后我把这个依赖关系添加到BaseRouter函数get和list中,并把结果传递给BaseObjectgetlist函数。
如承诺;一个完整的工作示例:

from typing import Optional, TypeVar, Type, Generic, Any, Union, Sequence
from fastapi import Depends, APIRouter, FastAPI
from pydantic import BaseModel

class ObjBase(BaseModel):
    name: Optional[str]

class ObjCreate(ObjBase):
    pass

class ObjUpdate(ObjBase):
    pass

class Obj(ObjBase):
    id: int

SchemaType = TypeVar("SchemaType", bound=BaseModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class Session:
    def __str__(self):
        return "I am a session!"

async def injecting_session():
    print("Creating Session")
    return Session()

class BaseService(Generic[SchemaType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[SchemaType]):
        self.model = model

    def get(self, id: Any, session: Session) -> Any:
        print(session)
        return {"id": id}

    def list(self, session: Session) -> Any:
        print(session)
        return [
            {"id": 1},
            {"id": 2},
        ]

class ObjService(BaseService[Obj, ObjCreate, ObjUpdate]):
    def __init__(self):
        super(ObjService, self).__init__(Obj)

def get_obj_service() -> ObjService:
    return ObjService()

SchemaType2 = TypeVar("SchemaType2", bound=BaseModel)
CreateSchemaType2 = TypeVar("CreateSchemaType2", bound=BaseModel)
UpdateSchemaType2 = TypeVar("UpdateSchemaType2", bound=BaseModel)

class BaseRouter(Generic[SchemaType2, CreateSchemaType2, UpdateSchemaType2]):
    def __init__(self, schema: Type[SchemaType2], prefix: str, service: BaseService):
        self.schema = schema
        self.service = service

        self.router = APIRouter(
            prefix=prefix
        )

        self.router.add_api_route("/", self.list, methods=['GET'])
        self.router.add_api_route("/{id}", self.get, methods=['GET'])

    def get(self, id, session=Depends(injecting_session)):
        return self.service.get(id, session)

    def list(self, session=Depends(injecting_session)):
        return self.service.list(session)

class ObjRouter(BaseRouter[Obj, ObjCreate, ObjUpdate]):
    def __init__(self, path, service):
        super(ObjRouter, self).__init__(Obj, path, service)

def get_obj_router(service=get_obj_service()) -> APIRouter:  # returns API router now
    return ObjRouter("/obj", service).router

app = FastAPI()
app.include_router(get_obj_router())

通过向injecting_session()添加参数,可以向使用依赖项的所有终结点添加参数。

相关问题