FastAPI CBV的实现

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

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 202,009评论 5 474
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 84,808评论 2 378
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 148,891评论 0 335
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,283评论 1 272
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,285评论 5 363
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,409评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,809评论 3 393
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,487评论 0 256
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,680评论 1 295
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,499评论 2 318
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,548评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,268评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,815评论 3 304
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,872评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,102评论 1 258
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,683评论 2 348
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,253评论 2 341