likes
comments
collection
share

FastApi（自用脚手架）+Snowy搭建后台管理系统（2）--脚手架应用APP抽象

作者站长头像
站长
· 阅读数 51

前言

由于我个人的封装一些插件存在注册时候有相关一定顺序约束的问题，特别是封装中存在一些依赖于中间件的插件，所以有必要进行APP初始化的时候一些流程的抽象，便于约束初始方法的调用。

为什么会存在顺序问题呢？主要是部分的插件比如一些中间件实现的方式有所不同，比如部分的中间件依赖于：

  • BaseHTTPMiddleware
  • 部分则是自定义的基于async def call(self, scope: Scope, receive: Receive, send: Send) -> None

而基于BaseHTTPMiddleware和自定义方式的__call__的中间件是实现的细节上是有所不同的，如果使用不恰当，或初始化顺序错乱的时候则会引发阻塞。

本小节主要主要

app应用初始化抽象类定义

为了约束初始化流程，所以我们需要定义一个抽象类，如下图所示结构：

FastApi（自用脚手架）+Snowy搭建后台管理系统（2）--脚手架应用APP抽象

app下的__init__.py对应抽象类的定义，具体代码如下所示：

import abc

from fastapi import FastAPI
import logging


class IApplicationBuilder():

    @classmethod
    @abc.abstractmethod
    def with_environment_settings(cls) -> 'IApplicationBuilder':
        raise NotImplemented

    @abc.abstractmethod
    def _instance_app(self) -> FastAPI:
        raise NotImplemented

    def _register_health_checks(self, app: FastAPI):
        pass

        @app.get("/health_checks", tags=['健康检查模块'], summary='健康检查接口')
        async def health_checks():
            return "ok"

    @abc.abstractmethod
    def _register_loguru_log_client(self, app: FastAPI) -> None:
        raise NotImplemented

    @abc.abstractmethod
    def _register_global_request(self, app: FastAPI) -> None:
        raise NotImplemented

    @abc.abstractmethod
    def _register_exception_handlers(self, app: FastAPI) -> None:
        raise NotImplemented

    @abc.abstractmethod
    def _register_plugins(self, app: FastAPI) -> None:
        raise NotImplemented

    @abc.abstractmethod
    def _register_routes(self, app: FastAPI) -> None:
        raise NotImplemented

    @abc.abstractmethod
    def _register_middlewares(self, app: FastAPI) -> None:
        raise NotImplemented

    def build(self) -> FastAPI:
        try:
            # 约束注册流程-避免错误
            logging.critical(f'约束注册流程')
            # 创建实例对象
            app = self._instance_app()
            # 执行错误注册
            self._register_exception_handlers(app)
            # 执行插件的注册----优先于路由注册，避免部分的全局对象加载问题
            self._register_plugins(app)
            # 执行中间件的注册
            self._register_middlewares(app)
            # 执行自定义的日志配置插件放在最后执行，以便获取到上下文的实例对象
            self._register_loguru_log_client(app)
            # 注册全局请求，最外层进行注册
            self._register_global_request(app)
            # 注册路由
            self._register_routes(app)
            # 健康检查路由
            self._register_health_checks(app)
            return app
        except Exception as e:
            logging.critical(f'项目启动失败:{e}')
            raise e

相关方法结构图如下所示：

FastApi（自用脚手架）+Snowy搭建后台管理系统（2）--脚手架应用APP抽象

对应的方法说明如下：

  • with_environment_settings 用于读取相关环境变量等配置信息
  • _instance_app 是完成对应的FastApi实例对象的创建
  • _register_health_checks 默认健康检查的一个API接口的注册
  • _register_loguru_log_client 初始化基于loguru日志插件
  • _register_global_request 初始化全局请求request实例对象（类似Flask中request）
  • _register_exception_handlers 初始化应用相关全局异常处理
  • _register_plugins 相关自定义插件的初始化
  • _register_routes 相关应用路由初始化
  • _register_middlewares 相关其他中间件初始化
  • build 是整个APP构建过程所需调用上面定义方法的实现

app应用抽象类实现示例

完成了对应抽象定义之后，则是对我们抽象进行具体的实现，如下代码所示：

FastApi（自用脚手架）+Snowy搭建后台管理系统（2）--脚手架应用APP抽象

在上图中我们的snowy_application.py中的FastApplicationBuilder是对IApplicationBuilder抽象类的实现，具体实现代码内容如下：

#!/usr/bin/evn python
# -*- coding: utf-8 -*-

from typing import Union

from fastapi import FastAPI, APIRouter
from starlette.middleware.cors import CORSMiddleware

from afastcore.middleware.disconnected import DisconnectedMiddleware
from afastcore.middleware.slash_strip_handle import SlashStripHandleMiddleware, SlashStripHandleMiddleware2
from afastcore.plugins.aiojobs import AiojobsPluginClient
from afastcore.plugins.async_cashews import AsyncCashewsPluginClient
from afastcore.plugins.cache_house import CacheHousePluginClient
from afastcore.plugins.db.sqlalchemy import SqlalchemyPluginForClassV2Client
from afastcore.plugins.pyee_event import EventEmitterPluginClient

from afastcore.plugins.sessions import SessionPluginClient
from afastcore.plugins.globalrequest import GlobalRequestPluginClient
from afastcore.plugins.loguru.client import LoguruPluginClient
from afastcore.plugins.swaggerui import SwaggeruiPluginClient
from afastcore.plugins.loguru import logger
from afastcore.app import IApplicationBuilder

from snowy_src.snowy_system.snowy_settings.development import DevSettings
from snowy_src.snowy_system.snowy_settings.production import ProSettings
from snowy_src.snowy_system.snowy_settings.setting import Environment


class FastApplicationBuilder(IApplicationBuilder):

    def __init__(self, *, settings: Union[DevSettings, ProSettings]):
        ''' 创建APP实例对象 '''
        if settings is None:
            # 检测是否存在配置实例对象
            raise RuntimeError('Must provide a valid Settings object')
        self.settings = settings

    @classmethod
    def with_environment_settings(cls) -> 'FastApplicationBuilder':
        ''' 根据当前环境变量选择指定的环境配置实例 '''
        return cls(settings=Environment.select())

    def _instance_app(self) -> FastAPI:
        # 创建实例对象
        return FastAPI(
            title=self.settings.project_name,
            version=self.settings.project_version,
            debug=self.settings.debug,
            # depends=[Depends(Logging)],
        )

    def _register_loguru_log_client(self, app: FastAPI) -> None:
        # 放在在最后处理因为是日志作用，所以一般使用的时候最后再执行注册
        pass
        # 日志插件初始化
        LoguruPluginClient(app=app,
                           settings=LoguruPluginClient.LoguruConfig(PROJECT_SLUG=self.settings.LOG_PROJECT_SLUG,
                                                                    FLITER_REQUEST_URL=self.settings.FLITER_REQUEST_URL,
                                                                    LOG_FILE_PATH=self.settings.LOG_FILE_PATH,
                                                                    MODEL=self.settings.LOG_MODEL)
                           )
        logger.info("LoguruPluginClient插件安装成功")

    def _register_global_request(self, app: FastAPI) -> None:
        # 注册应用全局请求的request
        pass
        GlobalRequestPluginClient(app=app)
        logger.info("GlobalRequestPluginClient插件安装成功")

    def _register_plugins(self, app: FastAPI) -> None:
        # 应用注册注册插件
        pass
        # 离线本地文档浏览
        SwaggeruiPluginClient(app=app, proxy=self.settings.swaggerui_proxy)
        # 数据库插件
        SqlalchemyPluginForClassV2Client(app=app, settings=SqlalchemyPluginForClassV2Client.SqlalchemySettings(
            MYSQL_SERVER_HOST=self.settings.MYSQL_SERVER_HOST,
            MYSQL_USER_NAME=self.settings.MYSQL_USER_NAME,
            MYSQL_PASSWORD=self.settings.MYSQL_PASSWORD,
            MYSQL_DB_NAME=self.settings.MYSQL_DB_NAME,
            SQLALCHEMY_DATABASE_ECHO=self.settings.SQLALCHEMY_DATABASE_ECHO,
            SQLALCHEMY_POOL_RECYCLE=self.settings.SQLALCHEMY_POOL_RECYCLE,
            SQLALCHEMY_POOL_PRE_PING=self.settings.SQLALCHEMY_POOL_PRE_PING,
            SQLALCHEMY_POOL_SIZE=self.settings.SQLALCHEMY_POOL_SIZE,
            SQLALCHEMY_MAX_OVERFLOW=self.settings.SQLALCHEMY_MAX_OVERFLOW
        ))
        # 会话插件
        SessionPluginClient(app=app, settings=SessionPluginClient.SessionConfig(
            is_auto_load=self.settings.session_is_auto_load
        ))
        # 缓存插件
        CacheHousePluginClient(app=app, settings=CacheHousePluginClient.CacheSettings(
            host='127.0.0.1', port=6379
        ))
        # 异步缓存插件
        AsyncCashewsPluginClient(app=app, settings=AsyncCashewsPluginClient.CacheSettings(
            url='redis://127.0.0.1/?db=1&socket_connect_timeout=0.5&safe=0'
        ))
        # 另一个后台任务的插件
        AiojobsPluginClient(app=app)
        # 类似信号事件分发插件
        EventEmitterPluginClient(app=app, settings=EventEmitterPluginClient.EventsSettings(
            events_name='events'
        ))

        # 性能检测数据，打印出相关调用链路的耗时处理
        # ProfilePluginClient(app=app, configs=ProfilePluginClient.ProfileConfig())
        # # 定时任务===需注意避免本地多worker情况启动多个可以加文件锁或其他锁
        # schedule = RocketrySchedulerPluginClient(app=app, configs=RocketrySchedulerPluginClient.SchedulerConfig())
        # @schedule.snowy_tasks('every 5 seconds')
        # async def do_things():
        #     print("定时任务！")
        #
        # @schedule.snowy_tasks(every('10 seconds', based="finish"))
        # async def do_permanently():
        #     "This runs for really long time"
        #     print(600000)
        #     await asyncio.sleep(600000)
        #
        # @schedule.snowy_tasks(every('2 seconds', based="finish"))
        # async def do_short():
        #     "This runs for short time"
        #     print(11111)
        #     await asyncio.sleep(1)

    def _register_exception_handlers(self, app: FastAPI) -> None:
        # 应用注册自定义的错误处理机制
        pass
        # 注册某个插件下特定的异常抛出处理
        # from snowy_src.snowy_system.snowy_common.snowy_errors import setup_snowy_ext_exception
        from snowy_src.snowy_common.snowy_errors.globalexption import setup_snowy_ext_exception
        setup_snowy_ext_exception(app=app)

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

        logger.info("路由开始注册")
        snowy_app_router = APIRouter(tags=['snowy后台管理系统模块'])
        # from snowy_src.snowy_business.snowy_modules.auth.setup import router_module as 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.auth.setup import GroupAPIRouterBuilder as auth
        # snowy_app_router.include_router(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)
        # 尾部斜杠重定向
        app.add_middleware(DisconnectedMiddleware)



    def _register_middlewares(self, app: FastAPI) -> None:
        pass
        app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )
        # 测试验证可以读取响应报文和读取请求报文信息的中间件
        from afastcore.middleware.request_response import RequestResponseMiddleware
        app.add_middleware(RequestResponseMiddleware)

# ############################################################################
# #############################APP对象应用构建 #############################
# ############################################################################
app = FastApplicationBuilder \
    .with_environment_settings() \
    .build()

FastApplicationBuilder对象说明

为什么需要在抽象类实现类中实例化FastApplicationBuilder对象实例，首先我们看一下相关应用启动图中两个文件：

snowy_application.py

snowy_launch.py

其中另一个snowy_launch.py是应用程序的入口文件，其代码如下所示：

#!/usr/bin/evn python
# -*- coding: utf-8 -*-

# 必须在外部载入app对象
from snowy_src.snowy_application import app

if __name__ == "__main__":
    # 使用os.path.basename函数获取了当前文件的名称，并将.py文件扩展名替换为空字符串\
    # import os
    # app_modeel_name = os.path.basename(__file__).replace(".py", "")
    from pathlib import Path
    # 使用Path函数获取了当前文件的名称，并将.py文件扩展名替换为空字符串\
    app_modeel_name = Path(__file__).name.replace(".py", "")
    import uvicorn
    # 使用uvicorn.run函数运行了一个应用程序。它指定了应用程序的主机和端口，并且设置了reload参数为True。
    uvicorn.run(f"{app_modeel_name}:app", host='127.0.0.1', port=31110, reload=True, workers=1)

才上面的代码可以看到我们的snowy_launch.py主要是通过uvicorn来运行我们的应用程序，在这个文件当中，我们直接导入在snowy_application.py实例化app对象。

from snowy_src.snowy_application import app

如此可以正常运行我们的应用程序，且不会重复的出现运行【两次问题】。

假如我们的把snowy_launch.py的代码放到snowy_application.py中的话，则会出现运行【两次问题】。如下代码说明所示：

# ############################################################################
# ############################# 启动说明 #############################
# ############################################################################
# 如果下面的代码放在main中的话，
# app = FastApplicationBuilder \
#     .with_environment_settings() \
#     .build()
# 也会类似下面的代码执行两次！
# if __name__ == "__main__":
#     直接在这里跑的会会运行两次，需要注意。所以建议单独放到main中运行，否则一些注册会执行多次
#     import uvicorn
#     import os
#     # 使用os.path.basename函数获取了当前文件的名称，并将.py文件扩展名替换为空字符串
#     app_modeel_name = os.path.basename(__file__).replace(".py", "")
#     # 使用uvicorn.run函数运行了一个应用程序。它指定了应用程序的主机和端口，并且设置了reload参数为True。
#     uvicorn.run(f"{app_modeel_name}:app", host='127.0.0.1',port=31100,reload=True)
# ############################################################################
# ############################# 启动说明 #############################
# ############################################################################

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

转载自:https://juejin.cn/post/7195383754418815034
评论
请登录