前言

在前面我们使用类的方式定义路由组的时是以回调的方式进行路由注册，看起来不是非常的优雅和以及缺乏可读性。如下图所示：

为了更具有优雅和可读性，我们可以进行改造一下。最终实现如下图所示的结果：

当前我们也有其他的方案进行改造为类的方式，比如fastapi-utils库中的cbv模式。如下代码所示：

from fastapi import Depends, FastAPI
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter


def get_x():
    return 10


app = FastAPI()
router = InferringRouter()  # Step 1: Create a router


@cbv(router)  # Step 2: Create and decorate a class to hold the endpoints
class Foo:
    # Step 3: Add dependencies as class attributes
    x: int = Depends(get_x)

    @router.get("/somewhere")
    def bar(self) -> int:
        # Step 4: Use `self.<dependency_name>` to access shared dependencies
        return self.x


app.include_router(router)

但是看cbv里面源码的时候，有点感觉复杂了，也不太容易理解，所以换了一个思路来实现。

改造过程：

要实现上面我们自己那种优雅的类定义路由方式，我们需要了解关键的原理过程。整体的核心思路就是基于APIRouter的实例对象的add_api_route方法来添加被标记为路由的相关端点函数。

1. 定义IBaseController基类

首先在基类中我们需要进行相关APIRouter的实例对象的创建，并且定义相关的当前路由组类的实例对象的创建，并返回当前APIRouter的实例对象。所以有以下的代码，如下图所示：

class IBaseController(metaclass=abc.ABCMeta):

    def __init__(self):
        self.api_router = None

    @property
    def cls(self):
        return type(self)

    @property
    def this_router_api(self):
        return self.api_router

    def _creat_api_router(self) -> APIRouter:
        pass
        if not self.api_router:
            pass
            self.api_router = APIRouter(
                prefix=getattr(self.cls, 'prefix'),
                tags=getattr(self.cls, 'tags'),
                dependencies=getattr(self.cls, 'dependencies'),
                default_response_class=getattr(self.cls, 'default_response_class'),
                responses=getattr(self.cls, 'responses'),
                callbacks=getattr(self.cls, 'callbacks'),
                routes=getattr(self.cls, 'routes'),
                redirect_slashes=getattr(self.cls, 'redirect_slashes'),
                default=getattr(self.cls, 'default'),
                dependency_overrides_provider=getattr(self.cls, 'dependency_overrides_provider'),
                route_class=getattr(self.cls, 'route_class'),
                on_startup=getattr(self.cls, 'on_startup'),
                on_shutdown=getattr(self.cls, 'on_shutdown'),
                deprecated=getattr(self.cls, 'deprecated'),
                include_in_schema=getattr(self.cls, 'include_in_schema'),
                generate_unique_id_function=getattr(self.cls, 'generate_unique_id_function'),
            )

        return self.api_router

    def _register_endpoint(self) -> APIRouter:
        # 获取当前注入的APIRouter对象
        assert hasattr(self, 'api_router'), '需要实例化APIRouter对象'
        # 获取当前注入的APIRouter对象
        self._creat_api_router()
        # 当前类下下定义对应的特定特点包含有被标记了_endpoint属性的路由函数
        for i in dir(self.cls):
            # 获取到注册到函数中的对应_endpoint函数对象
            route_endpoint = getattr(self.cls, i)
            # 判断是否是被RestRout给注册的端点函数
            if (inspect.isfunction(route_endpoint) or inspect.iscoroutinefunction(route_endpoint)) and hasattr(
                    route_endpoint, '_route_endpoint'):
                # 如果是指定的端点函数对象，透传当前self到当前的函数中
                # 要获取到对应的_endpoint绑定的函数，不能直接的获取getattr(self.cls, i)，不然会出现丢失__name__问题
                route_endpoint = getattr(route_endpoint, '_route_endpoint')
                # 路由参数对象信息
                route_args: RouteArgs = getattr(route_endpoint, '_route_args')
                # 传递self对象到对应_endpoint函数对象
                curr_route_endpoint = functools.partial(route_endpoint, self)
                # 注意事项---处理经过functools.partial后丢失__name__的问题
                # setattr(curr_route_endpoint, '__name__', route_endpoint.__name__)
                route_args.name = route_args.name or route_endpoint.__name__

                def cleandoc():
                    curr_route_endpoint.__doc__ = ''

                # 函数注释说明信息。文档显示描述问题处理
                route_args.description = route_args.description or inspect.cleandoc(
                    route_endpoint.__doc__ or '') or cleandoc()
                # 开始添加当前被被RestRout给注册的端点函数
                self.api_router.add_api_route(**route_args.dict(),
                                              endpoint=curr_route_endpoint)
        return self.api_router

    def build(self) -> APIRouter:
        #  注册点点路由
        return self._register_endpoint()

    @classmethod
    def instance(cls) -> APIRouter:
        """实例化"""
        return cls().build()

如上代码所示，build是创建实现注册被装饰绑定注解端点函数的注册和识别，内部调用的是自定义实现的_register_endpoint方法，在_register_endpoint方法内部中，主要关键的识别的思路是：

• 首先实例化APIRouter的实例对象
• 然后遍历该定义的所有的函数或协程函数
• 判断被遍历的函数是否包含有相关的_route_endpoint标签属性，如果有说明是它是对应端点路由函数。
• 然后获取对应的被标记为_route_endpoint标签属性的函数中包含有对应的【注入的端点函数中路由参数对象】【端点函数】
• 【获取到对应的端点函数】后，使用偏函数透传当前类的实例化对象self对象到【端点函数】内部
• 然后通过self.api_router.add_api_route进行添加对应的路由对象。

2. 定义待注入APIRouter对象参数

我们的APIRouter对象参数需要通过装饰器的方式注入到当前的cls中，然后可以通过对应的属性获取注入的参数项，装饰器代码如下所示：

def RestInjectAPIRouter(prefix: str = "",
                        tags: Optional[List[Union[str, Enum]]] = None,
                        dependencies: Optional[Sequence[params.Depends]] = None,
                        default_response_class: Type[Response] = Default(JSONResponse),
                        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
                        callbacks: Optional[List[BaseRoute]] = None,
                        routes: Optional[List[routing.BaseRoute]] = None,
                        redirect_slashes: bool = True,
                        default: Optional[ASGIApp] = None,
                        dependency_overrides_provider: Optional[Any] = None,
                        route_class: Type[APIRoute] = APIRoute,
                        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
                        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
                        deprecated: Optional[bool] = None,
                        include_in_schema: bool = True,
                        generate_unique_id_function: Callable[[APIRoute], str] = Default(
                            generate_unique_id
                        )):
    def back(cls):
        cls.prefix = prefix  # 给类添加属性
        cls.tags = tags  # 给类添加属性
        cls.redirect_slashes = redirect_slashes  # 给类添加属性
        cls.deprecated = deprecated  # 给类添加属性
        cls.include_in_schema = include_in_schema  # 给类添加属性
        cls.dependencies = dependencies  # 给类添加属性
        cls.default_response_class = default_response_class  # 给类添加属性
        cls.responses = responses  # 给类添加属性
        cls.callbacks = callbacks  # 给类添加属性
        cls.routes = routes  # 给类添加属性
        cls.default = default  # 给类添加属性
        cls.dependency_overrides_provider = dependency_overrides_provider  # 给类添加属性
        cls.route_class = route_class  # 给类添加属性
        cls.on_startup = on_startup  # 给类添加属性
        cls.on_shutdown = on_shutdown  # 给类添加属性
        cls.generate_unique_id_function = generate_unique_id_function  # 给类添加属性

        return cls

    return back

上面定义的各项参数主要是在实例化路由组对象的时候用到，如下代码所示：

def _creat_api_router(self) -> APIRouter:
    pass
    if not self.api_router:
        pass
        self.api_router = APIRouter(
            prefix=getattr(self.cls, 'prefix'),
            tags=getattr(self.cls, 'tags'),
            dependencies=getattr(self.cls, 'dependencies'),
            default_response_class=getattr(self.cls, 'default_response_class'),
            responses=getattr(self.cls, 'responses'),
            callbacks=getattr(self.cls, 'callbacks'),
            routes=getattr(self.cls, 'routes'),
            redirect_slashes=getattr(self.cls, 'redirect_slashes'),
            default=getattr(self.cls, 'default'),
            dependency_overrides_provider=getattr(self.cls, 'dependency_overrides_provider'),
            route_class=getattr(self.cls, 'route_class'),
            on_startup=getattr(self.cls, 'on_startup'),
            on_shutdown=getattr(self.cls, 'on_shutdown'),
            deprecated=getattr(self.cls, 'deprecated'),
            include_in_schema=getattr(self.cls, 'include_in_schema'),
            generate_unique_id_function=getattr(self.cls, 'generate_unique_id_function'),
        )

    return self.api_router

3. 定义端点路由参数注入装饰器

由于相关的路由包含了非常多的自定义参数，所以首先我们定义一个类，用来保存对应传入的参数项，如下代码所示：

class ApiRouteArgs(BaseModel):
    path: str
    response_model: Optional[Type[Any]] = None
    status_code: Optional[int] = None
    tags: Optional[List[str]] = None
    dependencies: Optional[Sequence[params.Depends]] = None
    summary: Optional[str] = None
    description: Optional[str] = None
    response_description: str = "Successful Response"
    responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None
    deprecated: Optional[bool] = None
    methods: Optional[Union[Set[str], List[str]]] = None
    operation_id: Optional[str] = None
    response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None
    response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None
    response_model_by_alias: bool = True
    response_model_exclude_unset: bool = False
    response_model_exclude_defaults: bool = False
    response_model_exclude_none: bool = False
    include_in_schema: bool = True
    response_class: Union[Type[Response], DefaultPlaceholder] = Default(
        JSONResponse
    )
    name: Optional[str] = None
    route_class_override: Optional[Type[APIRoute]] = None
    callbacks: Optional[List[Route]] = None
    openapi_extra: Optional[Dict[str, Any]] = None

    class Config:
        arbitrary_types_allowed = True

然后实现对应的注入参数装饰器，如下代码所示：

def RestRoute(path: str,
           methods: Optional[List[str]] = None,
           response_model: Optional[Type[Any]] = None,
           status_code: Optional[int] = None,
           tags: Optional[List[Union[str, Enum]]] = None,
           dependencies: Optional[Sequence[params.Depends]] = None,
           summary: Optional[str] = None,
           description: Optional[str] = None,
           response_description: str = "Successful Response",
           responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
           deprecated: Optional[bool] = None,
           operation_id: Optional[str] = None,
           response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
           response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
           response_model_by_alias: bool = True,
           response_model_exclude_unset: bool = False,
           response_model_exclude_defaults: bool = False,
           response_model_exclude_none: bool = False,
           include_in_schema: bool = True,
           response_class: Type[Response] = Default(JSONResponse),
           name: Optional[str] = None,
           callbacks: Optional[List[BaseRoute]] = None,
           openapi_extra: Optional[Dict[str, Any]] = None,
           **kwargs: Any):
    def call_func(fun: AnyCallable) -> Callable[[AnyCallable], AnyCallable]:
        route_args = ApiRouteArgs(path=path,
                               name=name,
                               status_code=status_code,
                               methods=methods,
                               tags=tags,
                               dependencies=dependencies,
                               description=description,
                               summary=summary,
                               response_description=response_description,
                               responses=responses,
                               deprecated=deprecated,
                               operation_id=operation_id,
                               response_model_include=response_model_include,
                               response_model_exclude=response_model_exclude,
                               response_model_by_alias=response_model_by_alias,
                               response_model_exclude_unset=response_model_exclude_unset,
                               response_model_exclude_defaults=response_model_exclude_defaults,
                               response_model_exclude_none=response_model_exclude_none,
                               include_in_schema=include_in_schema,
                               response_class=response_class,
                               callbacks=callbacks,
                               openapi_extra=openapi_extra,
                               response_model=response_model,
                               **kwargs)

      
        setattr(fun, '_route_args', route_args)
        setattr(fun, '_route_endpoint', fun)
        return fun

    return call_func

上述比较关键地方主要是：

• 对传入的参数进行ApiRouteArgs对象的实例化。
• 然后包装保存到被装饰的fun中，具体实现代码为： setattr(fun, '_route_args', route_args)。
• 然后把当前的函数标记为是一个_route_endpoint函数。具体实现代码为：setattr(fun, '_route_endpoint', fun) 通过上面的关键代码就实现了对应的被装饰函数标记和保存对应的对应路由参数信息。

基于上面的RestRoute方法，可以默认的扩展出来

• RestRouteGet
• RestRoutePost
• RestRoutePatch
• RestRoutePut
• RestRouteDelete

4. 使用相关装饰器

下面基于IBaseController的基类实现路由组对象，如下代码所示：

@RestInjectAPIRouter(prefix='/asas', tags=[' 路由组'])
class AGroupAPIRouterBuilder(IBaseController):

    @RestRouteGet(path='/ssss', summary="你好")
    def sdfsdfsd(self):
        '''
        ASDHJASHJDAH
        :return:
        '''
        return '我是谁1！！'

    @RestRouteDelete(path='/ssss2')
    async def sdfsdfsd2222(self, sm: AuthAccountPasswordLoginParam):
        import asyncio
        await asyncio.sleep(1)
        return '我是谁2！！'

6. 添加到根路由对象

完成我们的AGroupAPIRouterBuilder对象创建后，接下来就是添加到我们的根路由组，或其他路由组对象。如下代码所示：

from snowy_src.snowy_common.snowy_controller.xxxx import AGroupAPIRouterBuilder
snowy_app_router.include_router(AGroupAPIRouterBuilder.instance())

至此，关于一个简易版路由组类的相关介绍分享已完成！以上内容分享纯属个人经验，仅供参考！

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾

END

简书：www.jianshu.com/u/d6960089b…

掘金：juejin.cn/user/296393…

公众号：微信搜【程序员小钟同学】

新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822