前言

在snowy的项目，我们看它的API接口是使用类的形式来组织的，我们也可以参考类似的的方式来定义我们的一系列的路由组对象。如下图所示：

实现之后，我们最终实现的形式如下图所示：

路由组实现过程

通过类定义路由组的方式其实也有对应的好处，就是可以进行相关公共逻辑的复用。

1. 抽象定义路由的基类

在第一版中，由于考虑的比较简单，在实现路由组内部的对应的路由注册的时候，采取的是内部函数进行注册的方式进行实现。后期会再考虑优化。

首先我们需要一个公共注册内部的子路由的回调方法，便于在该方法中获取对应的APIRouter对象的引用，来完成调用其对应的方式进行路由注册。所以其抽象后的代码如下所示：

class IGroupBaseAPIRouter(ABC):

   @abstractmethod
   async def register_group_router(self, router: APIRouter) -> None:
       '''
        模块路由统一注册
       :param router:
       :return:
       '''
       raise NotImplemented

2. 定义具有Hook性质的APIRouter

部分的场景下，我们需要全局对当前路由组进行请求前和请求后的处理，所以我们有必要进行相关自定义APIRouter。 最终我们通过继承于APIRouter来实现，最终代码如下所示：

class HookRequestRoute(APIRoute):

   @abstractmethod
   async def before_request(self, apiroute: APIRoute, request: Request) -> None:
       """如果需要修改请求信息，可直接重写此方法"""
       raise NotImplemented

   @abstractmethod
   async def after_request(self, request: Request, response: Response) -> None:
       """请求后的处理【记录请求耗时等，注意这里没办法对响应结果进行处理】"""
       raise NotImplemented

   def get_route_handler(self) -> Callable:
       original_route_handler = super().get_route_handler()

       async def custom_route_handler(request: Request) -> Response:
           response_info = None
           try:
               await self.before_request(self, request=request)
               response_info = await original_route_handler(request)

               return response_info
           finally:
               response = None
               if response_info:
                   response = Response(content=response_info.body,
                                       status_code=response_info.status_code,
                                       headers=dict(response_info.headers)
                                       )
                   print("路由中响应报文：", response_info.body)
               await self.after_request(request=request, response=response)
               pass

       return custom_route_handler

3. 定义实现抽象类的路由组类

上面我们完成相关抽象类的定义和具有拦截性质的APIRouter，接下来，我们需要实现具体的路由组类，并实现对应的方法。 针对上面关于具有拦截性质的APIRouter，我们可以分两种形式来实现具体路由组类，一种是不需要拦截，一种是需要拦截的，代码如下所示：

class IHookReqGroupAPIRouterBuilder(IGroupBaseAPIRouter):
    path_prefix: str = None
    tags: Optional[List[Union[str, Enum]]] = None
    router: Optional[APIRouter] = None
    dependencies: Optional[Sequence[params.Depends]] = None
    redirect_slashes: bool = True
    deprecated: Optional[bool] = None
    include_in_schema: bool = True

    def __init__(
            self,
            _path_prefix: str = '',
            _tags: Optional[List[Union[str, Enum]]] = None,
            _dependencies: Optional[Sequence[params.Depends]] = None,
            _deprecated: Optional[bool] = None,
            *args, **kwargs,
    ):
        # 路由前缀信息的定义
        self._route_map = None
        self._path_prefix = self.path_prefix or _path_prefix
        # 路由组
        self._tags_fix = self.tags or _tags
        self._deprecated = self.deprecated or _deprecated
        # self._include_in_schema = self.include_in_schema or _include_in_schema
        self._router = self.router or APIRouter(prefix=self._path_prefix, tags=self._tags_fix,
                                                dependencies=self.dependencies or _dependencies,
                                                redirect_slashes=self.redirect_slashes ,
                                                include_in_schema=self.include_in_schema,
                                                deprecated=self._deprecated or _deprecated,

                                                *args, **kwargs)

    @abstractmethod
    async def before_request(self, request: Request, curr_apiroute: APIRoute) -> None:
        """如果需要修改请求信息，可直接重写此方法"""
        raise NotImplemented

    @abstractmethod
    async def after_request(self, request: Request, response: Response) -> None:
        """请求后的处理【记录请求耗时等，注意这里没办法对响应结果进行处理】"""
        raise NotImplemented

    def build(self) -> APIRouter:
        # 动态进行方法引用
        setattr(HookRequestRoute, 'before_request', self.before_request)
        setattr(HookRequestRoute, 'after_request', self.after_request)
        self._router.route_class = HookRequestRoute
        self.register_group_router(self._router)
        return self._router

    @classmethod
    def instance(cls, _path_prefix: str = '',
                 _tags: Optional[List[Union[str, Enum]]] = None,
                 _dependencies: Optional[Sequence[params.Depends]] = None,
                 *args, **kwargs) -> APIRouter:
        """实例化"""
        return cls(_path_prefix=_path_prefix,
                   _tags=_tags,
                   _dependencies=_dependencies,
                   *args, **kwargs
                   ).build()


class IGroupAPIRouterBuilder(IGroupBaseAPIRouter):
    path_prefix: str = None
    tags: Optional[List[Union[str, Enum]]] = None
    router: Optional[APIRouter] = None
    dependencies: Optional[Sequence[params.Depends]] = None
    redirect_slashes: bool = True
    deprecated: Optional[bool] = None
    include_in_schema: bool = True

    def __init__(
            self,
            _path_prefix: str = '',
            _tags: Optional[List[Union[str, Enum]]] = None,
            _dependencies: Optional[Sequence[params.Depends]] = None,
            _deprecated: Optional[bool] = None,
            *args, **kwargs,
    ):
        # 路由前缀信息的定义
        self._route_map = None
        self._path_prefix = self.path_prefix or _path_prefix
        # 路由组
        self._tags_fix = self.tags or _tags
        self._deprecated = self.deprecated or _deprecated
        self._router = self.router or APIRouter(prefix=self._path_prefix, tags=self._tags_fix,
                                                dependencies=self.dependencies or _dependencies,
                                                redirect_slashes=self.redirect_slashes ,
                                                include_in_schema=self.include_in_schema,
                                                deprecated=self._deprecated or _deprecated,

                                                *args, **kwargs)

    def build(self) -> APIRouter:
        # 动态进行方法引用
        self.register_group_router(self._router)
        return self._router

    @classmethod
    def instance(cls, _path_prefix: str = '',
                 _tags: Optional[List[Union[str, Enum]]] = None,
                 _dependencies: Optional[Sequence[params.Depends]] = None,
                 *args, **kwargs) -> APIRouter:
        """实例化"""
        return cls(_path_prefix=_path_prefix,
                   _tags=_tags,
                   _dependencies=_dependencies,
                   *args, **kwargs
                   ).build()

如上两个路由组类中，内部我们首先定义了一些路由组对象需要的参数信息，比如

• prefix 路由组对象前缀
• tags 路由组对象标签文档
• _dependencies 依赖项
• redirect_slashes 是否解决斜杠重定向问题
• deprecated 是否标记废弃API接口
• include_in_schema 是否显示在文档中

当在初始化的时候，我们会创建当前对应的APIRouter对象的实例。然后我们再定义相关build方法在这个地方，完成相关APIRouter对象回调传入我们上面定义好的抽象方法中，如下代码明细所示：

def build(self) -> APIRouter:
    # 动态进行方法引用
    setattr(HookRequestRoute, 'before_request', self.before_request)
    setattr(HookRequestRoute, 'after_request', self.after_request)
    self._router.route_class = HookRequestRoute
    self.register_group_router(self._router)
    return self._router

最终我们定义个instance实现相关的实例过程的封装：

@classmethod
def instance(cls, _path_prefix: str = '',
             _tags: Optional[List[Union[str, Enum]]] = None,
             _dependencies: Optional[Sequence[params.Depends]] = None,
             *args, **kwargs) -> APIRouter:
    """实例化"""
    return cls(_path_prefix=_path_prefix,
               _tags=_tags,
               _dependencies=_dependencies,
               *args, **kwargs
               ).build()

至此之后，我们后续就可以根据此方法来完成具体的路由组创建流程。

4.定义一些辅助装饰器

为了更接近于java那种定义的方式，也为了更好进行一些属性的分职责，所以定义一些辅助的装饰器，负责给要被装饰的类（路由组类）添加指定的一些参数，具体装饰器如下代码所示：

from enum import Enum
from typing import (
    List,
    Optional,
    Sequence,
    Union,
)

from fastapi import params
from starlette.routing import Mount as Mount  # noqa


def RestControllerApi(path_prefix: str = ""):
    def back(cls):
        cls.path_prefix = path_prefix  # 给类添加属性
        return cls

    return back


def RestControllerInject(dependencies: Optional[Sequence[params.Depends]] = None):
    def back(cls):
        cls.dependencies = dependencies  # 给类添加属性
        return cls

    return back


def RestDocumentApi(
        tags: Optional[List[Union[str, Enum]]] = None,
        redirect_slashes: bool = True,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True):
    def back(cls):
        cls.tags = tags  # 给类添加属性
        cls.redirect_slashes = redirect_slashes  # 给类添加属性
        cls.deprecated = deprecated  # 给类添加属性
        cls.include_in_schema = include_in_schema  # 给类添加属性
        return cls

    return back

5.具体路由组类实现

相关基础前置工作完成，就剩下具体路由组类了，首先我们看一下具体实现的代码，如下所示：

import asyncio
import time
from fastapi import Depends, Request, Response
from fastapi.routing import APIRoute
from afast_core.core_libs.group_router import IHookReqGroupAPIRouterBuilder
from afast_core.core_plugins.async_cashews import global_cashews
from snowy_src.snowy_common.snowy_controller import IBaseController
from snowy_src.snowy_common.snowy_controller.decorater import RestControllerApi, RestDocumentApi, RestControllerInject
from snowy_src.snowy_common.snowy_dependencies.inject_backgrond_tasks import BackgroundTasksDependencyInject
from snowy_src.snowy_system.snowy_modules.auth.controllers.doLogin import DoLoginController
from snowy_src.snowy_system.snowy_modules.auth.controllers.get_piccaptcha import GetPicCaptchaController
from snowy_src.snowy_system.snowy_modules.auth.tasks import newtasks


def module_dependencies(request: Request):
    print("模块下的全局依赖注入", request)


@RestControllerApi(path_prefix='/auth')
@RestDocumentApi(tags=['AUTH-b端模块'], include_in_schema=True)
@RestControllerInject(dependencies=[Depends(module_dependencies), Depends(newtasks)])
class GroupAPIRouterBuilder(IHookReqGroupAPIRouterBuilder):

    async def before_request(self, curr_apiroute: APIRoute, request: Request) -> None:
        """如果需要修改请求信息，可直接重写此方法"""
        pass

    async def after_request(self, request: Request, response: Response) -> None:
        """请求后的处理"""
        pass

    def register_group_router(self, router):
        '''
        API组路由的注册关联
        :param router:
        :return:
        '''

        # # 注册路由
        @router.get("/b/getPicCaptcha2/", summary='获取登入的验证码2')
        # 限流获取验证码--测试
        @global_cashews.rate_limit(key='my_rate_limit_key', period="1m", limit=10)
        async def getPicCaptcha(controller: IBaseController = Depends(GetPicCaptchaController)):
            '''
             获取验证码
            :param request:
            :return:
            '''
            return await controller.async_response()

        @router.post("/b/doLogin2/", summary='执行登入接口')
        async def doLogin(handel: IBaseController = Depends(DoLoginController)):
            '''
             输入账号登入
            :param request:
            :return:
            '''
            return await handel.async_response()

上面可以看到具体辅助装饰器的作用是对相关一些属性分组进制注册，这样也比较清晰。另外我们也实现具体的before_request和after_request路由拦截。然后在register_group_router中我们实现具体的路由注册的实现。

6. 最终文档效果

完成后，访问可视化API文档，如下图所示：

路由对象实现过程

从前面的代码我们可以看到，我们的路由对象是直接通过定义一个依赖项，我们知道每一个依赖项也可以使用类的方式来实现，所以我们的基于这种机制来实现我们的路由对象的“类化”定义，首先我们看一下具体的如下一个路由代码定义：

# # 注册路由
@router.get("/b/getPicCaptcha2/", summary='获取登入的验证码2')
# 限流获取验证码--测试
@global_cashews.rate_limit(key='my_rate_limit_key', period="1m", limit=10)
async def getPicCaptcha(controller: IBaseController = Depends(GetPicCaptchaController)):
    '''
     获取验证码
    :param request:
    :return:
    '''
    return await controller.async_response()

@router.post("/b/doLogin2/", summary='执行登入接口')
async def doLogin(handel: IBaseController = Depends(DoLoginController)):
    '''
     输入账号登入
    :param request:
    :return:
    '''
    return await handel.async_response()

从上面代码，我们可以看到我们统一通过IBaseController一个抽象类来统一的返回了响应报文的对象。 如下代码所示：

 return await controller.async_response()

1. IBaseController抽象定义

每一个IBaseController都是对每一个路由对象定义，如下抽象类定义代码所示：

class IBaseController(metaclass=abc.ABCMeta):
    def __init__(
            self,
            request: Request
    ):
        self._request: Request = request
        # 业务逻辑处理是否成功
        self.response_ok: bool = True
        # 提示信息
        self.response_msg: Union[str, None] = None

    #  # @async_property
    #  # 如果被async_property装饰则下面的判断asyncio.iscoroutinefunction(snowy_controller.response)会失效
    @property
    def request(self) -> Request:
        # 直接处理生成结果
        return self._request

    @abc.abstractmethod
    @overload
    async def business_login(self) -> Any:
        raise NotImplemented

    @abc.abstractmethod
    def business_login(self) -> Any:
        raise NotImplemented

    async def async_response(self):
        # 直接处理生成结果

        return SnowySuccess(data=await self.business_login(),
                            msg=self.response_msg) if self.response_ok else SnowyFail(
            msg=self.response_msg)

    def sync_response(self):
        return SnowySuccess(data=self.business_login(),
                            msg=self.response_msg) if self.response_ok else SnowyFail(
            msg=self.response_msg)
           
 class IAsyncSessionBaseController(IBaseController):
    def __init__(
            self,
            request: Request,
            async_session: Depends,
    ):
        super().__init__(request)
        self._async_session: AsyncSession = async_session

    @property
    def async_session(self) -> AsyncSession:
        # 直接处理生成结果
        return self._async_session
 

在上面的代码，

• request 参数是请求上下文请求对象，它本身也可以理解为是有个依赖项。
• business_login 方法是统一进行路由内的逻辑对象实现入口，其中需要注意的是我们定义了同步和异步两个方法。
• async_response 方法是统一返回business_login处理完成后的响应报文的结果。

2.抽象类的具体实现

基于抽象类之下，我们具体的实现，如下所示：

from fastapi import Request, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from afast_core.core_plugins.db.sqlalchemy import async_session_class_v2_dependency
from snowy_src.snowy_system.snowy_modules.auth.controllers.base import IAuthIBaseController
from snowy_src.snowy_system.snowy_modules.auth.enums import SaClientTypeEnum
from snowy_src.snowy_system.snowy_modules.auth.schemas import AuthAccountPasswordLoginParam


class IDoLoginControllerBaseController(IAuthIBaseController):
    '''
    定制相关约束检测函数逻辑步序
    '''
    pass


class IDoLoginController(IDoLoginControllerBaseController):
    '''
    定制相关约束检测函数逻辑步序
    '''

    def __init__(self, *, request: Request,
                 async_session: AsyncSession = Depends(async_session_class_v2_dependency),
                 schema_param: AuthAccountPasswordLoginParam,
                 ):
        super().__init__(request=request, async_session=async_session)
        # 入参参数
        self.schema_param = schema_param


class DoLoginController(IDoLoginController):
    async def business_login(self):
        # 把相关业务逻辑分到authService服务中处理
        return await self.authService.doLogin(self.schema_param, SaClientTypeEnum.B.value)

3. 服务层抽象封装和实现

如上的代码中，我们只需要在business_login实现具体的路由逻辑即可，其中，

return await self.authService.doLogin(self.schema_param, SaClientTypeEnum.B.value)

是对相关逻辑服务的封装，类似如下图所示的服务抽象：

最终我们的服务封装如下：

from abc import ABC, abstractmethod
from typing import Union, Tuple

from snowy_src.snowy_system.snowy_modules.auth.schemas import AuthAccountPasswordLoginParam, \
    AuthPhoneValidCodeLoginParam
from fastapi import Request

class IAuthService(ABC):
    """登录Service接口"""

    def __init__(self,request: Request=None):
        self.request= request

    @abstractmethod
    async def getPicCaptcha(self, type: str) -> Tuple[str, bytes]:
        '''
        根据不同的类型返回不同类型的验证编码，分为B端和C端的验证码
        :param type:指定用户登入的类型
        :return: 返回验证码字符串和图片base64编码
        '''
        pass

    @abstractmethod
    async def getPhoneValidCode(self, schema: AuthAccountPasswordLoginParam, type: str) -> str:
        '''
        根据不同的类型和登入的参数模型返回手机验证码
        :param schema: 用户登入的模型
        :param type: 指定用户登入的类型
        :return: 返回手机验证码
        '''
        pass

    @abstractmethod
    async def doLogin(self, schema: AuthAccountPasswordLoginParam, type: str) -> str:
        '''
        账号密码登录
        :param schema: 账号密码登入模型对象
        :param type:  指定用户登入的类型
        :return:
        '''
        pass

    @abstractmethod
    async def doLoginByPhone(self, schema: AuthPhoneValidCodeLoginParam, type: str) -> str:
        '''
        手机验证码登录
        :param schema:  手机验证码登录模型对象
        :param type: 指定用户登入的类型
        :return:
        '''
        pass

    @abstractmethod
    async def getLoginUser(self) -> str:
        '''
        获取B端登录用户信息
        :return:   返回是封装好的一个登入用户模型的详细信息
        '''
        pass

    @abstractmethod
    async def getClientLoginUser(self) -> str:
        '''
        获取C端登录用户信息
        :return:   返回是封装好的一个登入用户模型的详细信息
        '''
        pass

    @abstractmethod
    async def doLoginById(self, userId: str, device: str, type: str) -> str:
        '''
        根据用户id和客户端类型登录，用于第三方登录
        :return:   返回是封装好的一个登入用户模型的详细信息
        '''
        pass

对服务抽象类的实现，如下代码所示：

import base64
from typing import Tuple, Dict

import shortuuid
from fastapi import FastAPI, Request
from sqlalchemy.ext.asyncio import AsyncSession

from afast_core.core_libs.captcha.cilent import generate_captcha
from afast_core.core_tools.jwt_helper import SimpleAuth
from snowy_src.snowy_common.snowy_errors.enums import SnowySystemResultEnum
from snowy_src.snowy_common.snowy_errors.exception import SnowyBusinessException
from snowy_src.snowy_system.snowy_modules.auth.repository.user import SysUserCRUDRepository
from snowy_src.snowy_system.snowy_modules.auth.services.auth_base import IAuthService
from snowy_src.snowy_system.snowy_modules.auth.schemas import AuthAccountPasswordLoginParam, \
    AuthPhoneValidCodeLoginParam


class AuthService(IAuthService):
    pass

    def __init__(self, request: Request = None, async_session: AsyncSession = None):
        super().__init__(request)
        self.async_session = async_session

    async def getPicCaptcha(self, type: str = None) -> Dict[str, str]:
        '''
        根据不同的类型返回不同类型的验证编码，分为B端和C端的验证码
        :param type:指定用户登入的类型
        :return: 返回验证码字符串和图片base64编码
        '''
        pass
        random_str, image_data = generate_captcha(width=110, height=40)
        # 转换base64
        base64_data = str(base64.b64encode(image_data), "utf-8")
        # 写入session_code，不用redis
        session_code_reqno = shortuuid.uuid()

        # 设置其他相关的设置会话
        self.request.session["session_code"] = random_str
        self.request.session["session_code_reqno"] = session_code_reqno
        # 返回对应的 key值 表示分配当前的前端的key和验证码
        response_data = {
            'validCodeBase64': f"data:image/jpg;base64,{base64_data}",
            'validCodeReqNo': session_code_reqno
        }
        return response_data

    async def getPhoneValidCode(self, schema: AuthAccountPasswordLoginParam, type: str) -> str:
        '''
        根据不同的类型和登入的参数模型返回手机验证码
        :param schema: 用户登入的模型
        :param type: 指定用户登入的类型
        :return: 返回手机验证码
        '''
        pass

    async def doLogin(self, schema_param: AuthAccountPasswordLoginParam, type: str):
        '''
        账号密码登录
        :param schema: 账号密码登入模型对象
        :param type:  指定用户登入的类型
        :return:
        '''
        pass

        async def check_session_code():
            '''
            检测对应的会话ID信息
            :return:
            '''
            try:
                session_code = self.request.session["session_code"]
                session_code_reqno = self.request.session["session_code_reqno"]
            except Exception as e:
                raise SnowyBusinessException(msg='用户Cokkie被禁用！')
            if session_code_reqno != schema_param.validCodeReqNo.lower():
                raise SnowyBusinessException(msg='验证码请求码错误！')
            if session_code != schema_param.validCode.lower():
                raise SnowyBusinessException(msg='验证码错误！')

        # 查询用户信息
        async def check_user():
            '''
            查询用户信息
            :return:
            '''
            self.userinfo = await self.sysuser_repo.get_by_username(account=schema_param.account)
            if not self.userinfo:
                raise SnowyBusinessException(SnowySystemResultEnum.SNC10004)
                # 性别(ENABLE-正常,DISABLE-冻结)
            # 查询用户状态
            if self.userinfo.user_status == 'DISABLE':
                raise SnowyBusinessException(SnowySystemResultEnum.SNC10008)

        async def check_user_password():
            '''
            检测当前check_user_password
            :return:
            '''
            from afast_core.core_tools.encrypt_helper import md5
            # 加密处理后在进行对比处理
            if self.userinfo.password != md5(schema_param.password).upper():
                # 密码校验错误
                raise SnowyBusinessException(SnowySystemResultEnum.SNC10007)

        async def creat_user_token():
            '''
            检测当前creat_user_token
            :return:
            '''
            _date = {
                'account': self.userinfo.account,
                'account_id': self.userinfo.id,
                'org_id': self.userinfo.org_id,
                'position_id': self.userinfo.position_id,
            }
            # 生成对于的此用户信息的token, 创建授权码，并设置过期的时间--授权码有效期为 24 * 31天
            try:
                isok, token = SimpleAuth.create_token_by_data(sub='xiaozhongtongxue', data=_date,
                                                              exp_time=60 * 60 * 24 * 2)
            except Exception:
                raise SnowyBusinessException(SnowySystemResultEnum.SNC10010)

            return token

        # 创建当前请求业务逻辑需要相关的内部的变量信息，仅限于当前业务逻辑中
        # 逻辑判断流程
        self.userinfo = None
        # 依赖的数据层操作
        self.sysuser_repo = SysUserCRUDRepository(self.async_session)
        # ================================先检查检验验证输入胡session_code和session_code_reqno信息
        await check_session_code()
        # ================================查询用户信息
        await check_user()
        # ================================检测当前check_user_password
        await check_user_password()
        # 返回结果报文信息结果
        token = await creat_user_token()
        return token

    async def doLoginByPhone(self, schema: AuthPhoneValidCodeLoginParam, type: str) -> str:
        '''
        手机验证码登录
        :param schema:  手机验证码登录模型对象
        :param type: 指定用户登入的类型
        :return:
        '''
        pass

    async def getLoginUser(self) -> str:
        '''
        获取B端登录用户信息
        :return:   返回是封装好的一个登入用户模型的详细信息
        '''
        pass

    async def getClientLoginUser(self) -> str:
        '''
        获取C端登录用户信息
        :return:   返回是封装好的一个登入用户模型的详细信息
        '''
        pass

    async def doLoginById(self, userId: str, device: str, type: str) -> str:
        '''
        根据用户id和客户端类型登录，用于第三方登录
        :return:   返回是封装好的一个登入用户模型的详细信息
        '''
        pass

完成路由组和路由对象后，最终直接进行注入的根路由即可，如下代码所示：

def _register_routes(self, app: FastAPI) -> None:
    pass

    logger.info("路由开始注册")
    snowy_app_router = APIRouter(tags=['snowy后台管理系统模块'])
    # from snowy_src.snowy_business.snowy_modules.core_auth.setup import router_module as core_auth
    # from snowy_src.snowy_business.snowy_modules.dev.setup import router_module as dev
    # from src.modules.snowy.modules.sys import router_module as sys
    # from src.modules.snowy.modules.dev import router_module as dev
    # from snowy_src.snowy_system.snowy_modules.core_auth.setup import GroupAPIRouterBuilder as core_auth
    # snowy_app_router.include_router(core_auth.instance())

    from snowy_src.snowy_system.snowy_modules.auth.setup import GroupAPIRouterBuilder as auth2
    snowy_app_router.include_router(auth2.instance())

    # snowy_app.include_router(dev)
    logger.info("路由模块插件导入插件安装成功")
    # 加入模块路由组
    app.include_router(snowy_app_router)

至此，关于路由组和路由对象相关介绍分享已完成！以上内容分享纯属个人经验，仅供参考！文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾