幂简集成LOGO
API Hub
找服务商
API知识库
我是服务商
产品和服务
关于我们
控制台
所有文章 > API产品 > FastAPI “类视图”管理接口
FastAPI “类视图”管理接口

FastAPI “类视图”管理接口

作者: weixin01 2024-12-13 

这里的类视图和常看到的Python Web中的类视图不太一致（常见的类视图仅提供了POST,GET,DELETE, PUT,...等http方法），这里的类视图时将原本的router.get的操作用类来统一管理， 通过装饰器实现。
from fastapi import APIRouter



router = APIRouter()



@router.get("/")

def index():

    return "/index"

@router.get 背后

访问起源码可以看见调到self.add_api_route传了一个func，这个func在上面的代码中就是index这个函数 / endpoint，到这一层就够了
通过上面得知每个router.xx后面都调用了add_api_route，那要实现类视图，需要步骤如下：1. 创建APIRouter示例，2. 将类中的某些方法add_api_route加到APIRouter实例中即可

Controller 装饰器

这里简化了参数描述，会缺少代码提示，实际和APIRouter的参数定义一致就行

class Controller:



    def __init__(self, **kwargs):

        """kwargs 等同于APIRouter 实例化入参"""

        self.kwargs = kwargs



    def __call__(self, cls):

        # 创建router实例

        router: APIRouter = APIRouter(

                                      **self.kwargs

                                      )

        # 返回被装饰类的所有方法和属性名称

        for attr_name in dir(cls):

            # 通过反射拿到对应属性的值 或方法对象本身

            attr = getattr(cls, attr_name)

            # 简单处理，如果函数返回的是个字典并且endpoint在这里面，就添加到router上

            if isinstance(attr, dict) and "endpoint" in attr:

                router.add_api_route(**attr)

        # 然后把router 给被装饰的类上，再返回被装饰类

        cls.router = router

        return cls

RequestMapping 装饰器

被装饰方法返回带endpoint的字典，让Controller能够扫描到

class RequestMapping:

    """请求"""



    def __init__(self, **kwargs):

        self.kwargs = kwargs



    def __call__(self, func):

        # 这里这个endpoint 对应的value 就是被装饰的函数

        # 返回的内容其实是符合self.api_add_route的入参要求

        return {"endpoint": func, **self.kwargs}

测试代码 main.py

from fastapi import APIRouter



class Controller:



    def __init__(self, **kwargs):

        """kwargs 等同于APIRouter 实例化入参"""

        self.kwargs = kwargs



    def __call__(self, cls):

        # 创建router实例

        router: APIRouter = APIRouter(

            **self.kwargs

        )

        # 返回被装饰类的所有方法和属性名称

        for attr_name in dir(cls):

            # 通过反射拿到对应属性的值 或方法对象本身

            attr = getattr(cls, attr_name)

            # 简单处理，如果函数返回的是个字典并且endpoint在这里面，就添加到router上

            if isinstance(attr, dict) and "endpoint" in attr:

                router.add_api_route(**attr)

        # 然后把router 给被装饰的类上，再返回被装饰类

        cls.router = router

        return cls



class RequestMapping:

    """请求"""



    def __init__(self, **kwargs):

        self.kwargs = kwargs



    def __call__(self, func):

        # 这里这个endpoint 对应的value 就是被装饰的函数

        # 返回的内容其实是符合self.api_add_route的入参要求

        return {"endpoint": func, **self.kwargs}



def get_db():

    print("模拟db session")

    return "模拟db session"



@Controller(prefix="/demo", tags=["demo"])

class Demo:



    name: str = "ggbond"

    db_session: str = Depends(get_db)



    @RequestMapping(path="")

    def get_list(self):

        return [self.name, self.db_session]



from fastapi import FastAPI



app = FastAPI()



app.include_router(Demo.router)



if __name__ == '__main__':

    import uvicorn



    uvicorn.run(app)

问题

遇到的这些问题，解决方法其实之前大佬都写好了 https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py

将self识别成了查询参数

解决方法 update_endpoint_signature

将self这个方法签名修改成Depends(cls)– 实例化当前类对象，并依赖注入（依赖注入的参数，无需用户传递哦）。

def update_endpoint_signature(endpoint, cls_obj):

    """

    source：https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py#L92

    修改端点签名，将self参数改成关键字参数 self = Depends(cls)

    :param endpoint: 端点

    :param cls_obj: 被装饰的类 本身

    :return:

    """

    # 获取原始端点函数的签名

    old_signature = inspect.signature(endpoint)

    # 获取参数列表

    old_parameters = list(old_signature.parameters.values())

    # 获取原始端点函数的第一个参数， self

    old_first_parameter = old_parameters[0]

    # 将第一个参数替换为依赖注入 当前类 的参数

    # 

    new_first_parameter = old_first_parameter.replace(default=Depends(cls_obj))

    new_parameters = [new_first_parameter] + [

        parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY)

        for parameter in old_parameters[1:]

    ]  # 替换其他参数的类型为 KEYWORD_ONLY，以便正确执行依赖注入

    new_signature = old_signature.replace(parameters=new_parameters)  # 创建新的签名

    setattr(endpoint, "__signature__", new_signature)  # 替换端点函数的签名

依赖注入未被正确解析

解决方法

需要修改类的签名、和init方法，并在init方法时调用依赖注入的函数这里的实例get_db，并把结果更新到db_session中去

CBV_CLASS_KEY = "__cbv_class__"



def _init_cbv(cls: type[Any]) -> None:

    """

    Idempotently modifies the provided cls, performing the following modifications:
    * The __init__ function is updated to set any class-annotated dependencies as instance attributes
    * The __signature__ attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
    """
    # 实例化之后就不用实例化了
    if getattr(cls, CBV_CLASS_KEY, False):  # pragma: no cover
        return  # Already initialized
    # 保留原本的 init
    old_init: Callable[..., Any] = cls.__init__
    # 获取原本的 实例签名
    old_signature = inspect.signature(old_init)
    # 删除self
    old_parameters = list(old_signature.parameters.values())[1:]  # drop self parameter
    # 改成关键字参数
    new_parameters = [
        x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
    ]
    # 依赖注入列表
    dependency_names: list[str] = []
    # 返回类 cls 中的所有类型标注，
    for name, hint in get_type_hints(cls).items():
        if is_classvar(hint):
            continue
        parameter_kwargs = {"default": getattr(cls, name, Ellipsis)}
        dependency_names.append(name)
        # 加入到关键字参数中
        new_parameters.append(
            inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
        )
    # 替换签名参数
    new_signature = old_signature.replace(parameters=new_parameters)

    # 实例化时候的处理
    def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
        for dep_name in dependency_names:
            # pop 会拿到依赖注入的返回值
            dep_value = kwargs.pop(dep_name)
            # 设置成属性
            setattr(self, dep_name, dep_value)
        # 再调用原来的实例化方法
        old_init(self, *args, **kwargs)

    setattr(cls, "__signature__", new_signature)
    setattr(cls, "__init__", new_init)

完整实现

boot.py

"""

Source: https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py

"""



from __future__ import annotations



import inspect

from collections.abc import Callable

from typing import Any, TypeVar, get_type_hints



from fastapi import APIRouter, Depends

from pydantic.v1.typing import is_classvar



T = TypeVar("T")



CBV_CLASS_KEY = "__cbv_class__"



def _init_cbv(cls: type[Any]) -> None:

    """

    Idempotently modifies the provided cls, performing the following modifications:
    * The __init__ function is updated to set any class-annotated dependencies as instance attributes
    * The __signature__ attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
    """
    if getattr(cls, CBV_CLASS_KEY, False):  # pragma: no cover
        return  # Already initialized
    old_init: Callable[..., Any] = cls.__init__
    old_signature = inspect.signature(old_init)
    old_parameters = list(old_signature.parameters.values())[1:]  # drop self parameter
    new_parameters = [
        x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
    ]
    dependency_names: list[str] = []
    for name, hint in get_type_hints(cls).items():
        if is_classvar(hint):
            continue
        parameter_kwargs = {"default": getattr(cls, name, Ellipsis)}
        dependency_names.append(name)
        new_parameters.append(
            inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
        )
    new_signature = old_signature.replace(parameters=new_parameters)

    def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
        for dep_name in dependency_names:
            dep_value = kwargs.pop(dep_name)
            setattr(self, dep_name, dep_value)
        old_init(self, *args, **kwargs)

    setattr(cls, "__signature__", new_signature)
    setattr(cls, "__init__", new_init)
    setattr(cls, CBV_CLASS_KEY, True)

def _update_cbv_route_endpoint_signature(cls: type[Any], endpoint: Callable[..., Any]) -> None:
    """
    Fixes the endpoint signature for a cbv route to ensure FastAPI performs dependency injection properly.
    """
    old_signature = inspect.signature(endpoint)
    old_parameters: list[inspect.Parameter] = list(old_signature.parameters.values())
    old_first_parameter = old_parameters[0]
    new_first_parameter = old_first_parameter.replace(default=Depends(cls))
    new_parameters = [new_first_parameter] + [
        parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY) for parameter in old_parameters[1:]
    ]
    new_signature = old_signature.replace(parameters=new_parameters)
    setattr(endpoint, "__signature__", new_signature)

class Controller:

    def __init__(self, **kwargs):
        """kwargs 等同于APIRouter 实例化入参"""
        self.kwargs = kwargs

    def __call__(self, cls):
        _init_cbv(cls)
        # 创建router实例
        router: APIRouter = APIRouter(
            **self.kwargs
        )
        # 返回被装饰类的所有方法和属性名称
        for attr_name in dir(cls):
            # 通过反射拿到对应属性的值 或方法对象本身
            attr = getattr(cls, attr_name)
            # 添加到router上
            if isinstance(attr, BaseRoute) and hasattr(attr, "kwargs"):
                _update_cbv_route_endpoint_signature(cls, attr.kwargs["endpoint"])

                if isinstance(attr, RequestMapping):
                    router.add_api_route(**attr.kwargs)
                elif isinstance(attr, WebSocket):
                    router.add_websocket_route(**attr.kwargs)
                else:
                    assert False, "Cls Type is RequestMapping or WebSocket"

        cls.router = router
        return cls

class BaseRoute:
    pass

class RequestMapping(BaseRoute):
    """请求"""

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    def __call__(self, func):
        # 这里这个endpoint 对应的value 就是被装饰的函数
        # 返回的内容其实是符合self.api_add_route的入参要求
        self.kwargs["endpoint"] = func
        return self

class WebSocket(BaseRoute):

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    def __call__(self, func):
        # 这里这个endpoint 对应的value 就是被装饰的函数
        # 返回的内容其实是符合self.api_add_route的入参要求
        self.kwargs["endpoint"] = func
        return self

main.py

from fastapi import Depends



from boot import Controller, RequestMapping, WebSocket



def get_db():

    print("模拟db session")

    return "模拟db session"



@Controller(prefix="/demo", tags=["demo"])

class Demo:

    name = "ggbond"

    # 带类型标注类属性会被加入到方法中如router.get(db_session: str = Depends(get_db))

    db_session: str = Depends(get_db)



    @RequestMapping(path="")

    def get_list(self, age: int):

        return [self.name, self.db_session]



    @WebSocket(path="/ws")

    async def ws(self):

        pass



from fastapi import FastAPI



app = FastAPI()



app.include_router(Demo.router)



if __name__ == '__main__':

    import uvicorn



    uvicorn.run(app)

文章转自微信公众号@7y记

上一篇：

FastAPI：重燃Python Web开发的火花

下一篇：

FastAPI，一个快速开发 API 的 Python 框架！
#你可能也喜欢这些API文章!

我们有何不同？

API服务商零注册

多API并行试用

数据驱动选型，提升决策效率

查看全部API→
🔥

热门场景实测，选对API

#AI文本生成大模型API

对比大模型API的内容创意新颖性、情感共鸣力、商业转化潜力

25个渠道
一键对比试用API 限时免费

#AI深度推理大模型API

对比大模型API的逻辑推理准确性、分析深度、可视化建议合理性

10个渠道
一键对比试用API 限时免费
内容目录
  1. @router.get 背后
  2. Controller 装饰器
  3. RequestMapping 装饰器
  4. 测试代码 main.py
  5. 问题
  6. 将self识别成了查询参数
  7. 依赖注入未被正确解析
  8. 完整实现