我们的Python服务器开发脚手架开放了

news2024/12/23 10:21:23

pdServer是我们实践过程中的产物，当我们在开发各类python应用时，经常需要一个server来提供服务。于是我们使用fastApi来提实现，并在这个过程中不断的完善，实现了JWT\SQL等。

我们的脚手架项目可以：

fastApi实现一个服务器；
PyQt5实现了GUI开发；
简单的SQL访问（未实现ORM）；
JWT实现的token鉴权；
logger日志组件；
ini系统配置持久化；
多线程(GUI/SERVER)与业务任务队列多协程实现并发；

怎么样，一个轻量服务需要的功能模块都有了，来看看是否符合需求吧！

先看效果再说过程

我们的脚手架项目，UI上包含两个：系统运行日志和系统参数设置。如果你想开发一个DashBoard统计一些运行数据，也是非常的简单。系统配置我们持久化到了本地ini文件中，因此，需要我们复制给出路径，打开config.ini文件进行配置。

（界面截图）

（config.ini截图）

代码结构

代码结构如下图所示，我们从上向下作一个简单介绍。

（代码结构）

GUI是我们的界面类

loader是启动器，mainWindow是主窗口，其它的是各子窗口，本项目的主窗口是支持多tab的。

business包是我们的业务处理包，这里是我们具体的业务实现，你可以按照自己的业务来命名。
PdBaseKits是我的基础工具包

这里实现了exceptions异常定义，日志配置，sql访问，token工具，以及一些常用的时间、文件方法。

server包是我们的fastApi包

在这个包中实现了服务启动，路由(router)，任务队列(queue)，model定义(schema)。

主流程介绍

这里我将简单的介绍几个主要的类，这涵养了启动过程与服务路由，更多详细的开发指南，我将在以后的文章中更新。

serverMain.py是我们的入口

if __name__ == '__main__':    setupLogger("kling")    logging.info("开始初始化系统配置文件……")
    initialSysConfig()    logging.info("完成初始化数据加载")
    setupWindow()    logging.info("完成界面加载")

从这里可以看到，其主要完成了系统的初始化。ini首先会检查是否在本地存在config.ini文件，如果不存在会新建，并使用缺省配置。

setupWindow来自loader.py，其完成GUI的加载。

    app = QApplication(sys.argv)    window = KlingServerGUI()    window.show()    sys.exit(app.exec())

这里调用了KlingServerGUI，这是我们的主窗口。

class KlingServerGUI(QMainWindow):    newStockHqData = ""    newStateData = ""    existTabName = []    MENU_RUN_STATE = "RUNSTATE"    MENU_SYS_SETTING = "HYFX"
    exeTimeArr = ["09"]
    # 初始化各个句柄    def __init__(self, parent=None):        super(KlingServerGUI, self).__init__(parent)
        self.showTabWidget = None        self.runStateBoxShow = None        self.SysSettingBoxShow = None
        self.existTabName = []        # 设置窗口标题        self.setWindowTitle('KlingApiServer(' + version + ')，请匆关闭！！')        # 设置窗口大小        self.resize(400, 600)        self.setWindowIcon(QIcon(":/logo.png"))        self.createMenu()        self.iniWelcomeTab()

我们会在这里初始化菜单，子窗口。

当我们在界面中点击“启动服务器”后，会启动我们的服务。

我们在router.py中可以添加我们的路由。

@router.post("/login", response_model=UserLoginResponse)async def login(body: User):    logging.info("login ["+body.user+"],["+body.pwd+"]")    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    access_token = create_access_token(data={"userName": body.user,'id':12}, expires_delta=access_token_expires)
    return {"user": 'zyy', "id": "12", 'token': access_token, 'nickname': 'dean', 'headUrl': 'i.png'}

这里我例出了login接口，其会返回token，在其它的接口中，你可以实现自己的业务代码。