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)
的反射能力失效,因为class的endpoint是不明确的,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,或者处理不妥之处,还请见谅