FastAPI 解读 by Gascognya程序员

FastAPI CBV的实现

2020-10-07  本文已影响0人  Gascognya

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,或者处理不妥之处,还请见谅

上一篇下一篇

猜你喜欢

热点阅读