FastAPI CBV的实现

作者: Gascognya | 来源:发表于2020-10-07 09:22 被阅读0次

    FastAPI天生不支持CBV

    Starlette层面上,还提供有CBV的支持,但是在FastAPI的实现,都是默认不考虑CBV的。
    fastapi中实现原生cbv,需要涉及到源码的大量修改。因为fastapi的路由对应的app。包含了endpoint,同时也包含了对endpoint的依赖解析(dependant)。
    当APIRoute节点生成时,会拿着传进来的endpoint,通过get_dependant()解析endpoint,并生成一份特定的依赖树。这是FastAPI感知力的关键。有了这份依赖,Fastapi才能像静态语言那样,对数据进行强制要求(因为它十分清楚该endpoint需要哪些东西)。

    为什么不能是cbv

    我们上面说到,一份endpoint有一份专属的dependant,才能使fastapi的机能正常工作,所以我们为APIRoute提供的endpoint,一定要为function形式才行。因为class具有多个方法,如果直接将class作为app(这在starlette中是允许的),就算可以通过__call__解决一些问题,但是会导致诸如inspect.signature(call)的反射能力失效,因为classendpoint是不明确的,fastapi不清楚应该生成哪个方法的dependant。所以除非我们从源码层面解决,生成新的专属解决方案。否则,传入APIRoute时就必须是function形式。

    那么该怎么做?

    虽然不能像Starlette那样,将类作为app,传入scope再进行dispatch。但是我们可以再形式上接近cbv。即编写时仍然是cbv形式。但实际逻辑是将cbv的方法拆出来,当做四个不同的endpoint生成路由。再使用层面上仍然可以达到近似效果。

    下面提供一个简单的思路

    class CbvMeta(type):
    
        def __new__(mcs, *args, **kwargs):
            cls = super().__new__(mcs, *args, **kwargs)
            if cls.__name__ != 'CbvTest':
                asgi = getattr(cls, 'app', None)
                path = getattr(cls, 'path', None)
    
                assert asgi is not None and isinstance(asgi, FastAPI), f"请为{cls.__name__}类配置正确的ASGI应用"
                assert path is not None, f"请为{cls.__name__}类配置正确的路径"
    
                for http_method in ['get', 'post', 'put', 'delete', 'options', 'head', 'patch', 'trace']:
    
                    def parse(item):
                        item = getattr(cls, item)
    
                        return item[http_method] if isinstance(item, MethodDict) else None
    
                    if method := getattr(cls, http_method, None):
                        getattr(asgi, http_method, None)(
                            path=path,
                            tags=parse('tags') or [cls.__name__],
                            summary=parse('summary') or f'{cls.__name__}.{http_method}',
                            operation_id=parse('operation_id') or f'cbv_{path[1:-2]}_{http_method}',
                            response_model=parse('response_model'),
                            status_code=parse('status_code'),
                            dependencies=parse('dependencies'),
                            description=parse('description'),
                            response_description=parse('response_description'),
                            responses=parse('responses'),
                            deprecated=parse('deprecated'),
                            response_model_include=parse('response_model_include'),
                            response_model_exclude=parse('response_model_exclude'),
                            response_model_by_alias=parse('response_model_by_alias'),
                            response_model_exclude_unset=parse('response_model_exclude_unset'),
                            response_model_exclude_defaults=parse('response_model_exclude_defaults'),
                            response_model_exclude_none=parse('response_model_exclude_none'),
                            include_in_schema=parse('include_in_schema'),
                            response_class=parse('response_class'),
                            name=parse('name'),
                            callbacks=parse('callbacks'),
                        )(method)
            return cls
    
    
    class MethodDict:
        def __init__(self, get=None, post=None, put=None, delete=None, options=None, head=None, patch=None, trace=None,
                     default=None):
            self.get = get if get is not None else default
            self.post = post if post is not None else default
            self.put = put if put is not None else default
            self.delete = delete if delete is not None else default
            self.options = options if options is not None else default
            self.head = head if head is not None else default
            self.patch = patch if patch is not None else default
            self.trace = trace if trace is not None else default
            self.default = default
    
        def __getitem__(self, item):
            return getattr(self, item, self.default)
    
    
    class CbvTest(metaclass=CbvMeta):
        app: FastAPI = None
        path: str = None
    
        tags: Optional[List[str]] = MethodDict(default=None)
        summary: Optional[str] = MethodDict(default=None)
        operation_id: Optional[str] = MethodDict(default=None)
        response_model: Optional[Type[Any]] = MethodDict(default=None)
        status_code: int = MethodDict(default=200)
        dependencies: Optional[Sequence[params.Depends]] = MethodDict(default=None)
        description: Optional[str] = MethodDict(default=None)
        response_description: str = MethodDict(default="Successful Response")
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = MethodDict(default=None)
        deprecated: Optional[bool] = MethodDict(default=None)
        response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = MethodDict(default=None)
        response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = MethodDict(default=None)
        response_model_by_alias: bool = MethodDict(default=True)
        response_model_exclude_unset: bool = MethodDict(default=False)
        response_model_exclude_defaults: bool = MethodDict(default=False)
        response_model_exclude_none: bool = MethodDict(default=False)
        include_in_schema: bool = MethodDict(default=True)
        response_class: Optional[Type[Response]] = MethodDict(default=None)
        name: Optional[str] = MethodDict(default=None)
        callbacks: Optional[List[APIRoute]] = MethodDict(default=None)
    
    

    CBV的三个重点,一是通过面向对象使其更方便,二是方便同类不同方法的整合,三是可以方便不同方法共用资源。
    本示例使用metaclass的方式,将方法拆出来,手动调用app的装饰器。

    from cbv import CbvTest, MethodDict
    from fastapi import FastAPI
    
    app = FastAPI()
    
    
    class Test(CbvTest):
        app = app
        path = '/test1/'
    
        description = MethodDict(get='get方法', post='post方法', default=None)
        status_code = MethodDict(get=200, default=200)
        # ------------------------
        num = 1
    
        @classmethod
        def get(cls):
            return {'msg': cls.num}
    
        @classmethod
        def post(cls, item_id: int):
            return items[cls.num]
    
    
    items = {
        1: {"name": "Foo", "price": 50.2},
        2: {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
        3: {
            "name": "Baz",
            "description": "There goes my baz",
            "price": 50.2,
            "tax": 10.5,
        },
    }
    
    if __name__ == '__main__':
        import uvicorn
        uvicorn.run(app, host="127.0.0.1", port=8000)
    
    

    使用方法
    可以使用类方法作为endpoint,这样可以使用类的资源。
    在类属性中可以重写MethodDict,这里没有采用{‘get’:..., 'post':...}的字典形式,而是封装成了类。

    仅做示例用途,可能包含一些未知的bug,或者处理不妥之处,还请见谅

    相关文章

      网友评论

        本文标题:FastAPI CBV的实现

        本文链接:https://www.haomeiwen.com/subject/hfhcsktx.html