掩盖智能拱逗1、FastAPI实现主从表数据接口和SqlAlchemy的数据处理
WxPython跨平台开发框架全面采用Python语言进行构建,涵盖后端内容,使用基于SqlAlchemy+Pydantic+FastApi的后端架构。FastAPI启动之后,访问Swagger页面如下所示,列出了各业务表的相关接口。
以上为常规化接口,包括单一获取、ID验证、条件检索、分页检索、数量查询、添加、移除、更新等标准化接口,这些基础类接口通常封装在API的控制器基类中。
该后端接口采纳统一的接口协议,标准格式如下所示。
{
"success": false,
"result": T ,
"targetUrl": "string",
"UnAuthorizedRequest": false,
"errorInfo": {
"code": 0,
"message": "string",
"details": "string"
}
}
常规化接口与泛型结合使用,如此定义可以有效地抽象不同的业务类接口至基类控制器中。下面是FastApi后端的基类控制器定义。
[此处为图片1]有了上述基类定义好的常规化接口,子类只需继承基类控制器即可获得强大的功能接口。
通常API控制器的子类仅需继承基类,额外增加的接口按照常规化设计函数处理即可,可以参照基类的方式编写各种(GET、PUT、DELETE、POST)的处理函数。
[此处为图片1]我们来审视产品报价单和明细记录的处理,这两个代表不同的业务表,可以分开管理,通过限制它们的记录关系实现主从表灵活性的管理。
例如对于主表,在基类接口外定义额外两个方法,主要是删除主表时同时移除明细记录的关联操作。
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Path, Body
..................
# 创建路由以处理自定义接口
router = APIRouter()
@router.get(
"byorderno/{orderno}",
response_model=AjaxResponse[QuotationDto | None],
summary="根据报价单编号获取对象信息",
dependencies=[DependsJwtAuth],
)
async def find_by_orderno(
orderno: Annotated[str | None, Path(description="订单编号")],
request: Request,
db: AsyncSession = Depends(get_db),
):
# ip = await get_request_ip(request)
item = await quotation_crud.get_by_column(db, "handno", orderno)
item = QuotationDto.model_validate(item) if item else None
return AjaxResponse(item)
@router.delete(
"quotation-related/{id}",
response_model=AjaxResponse[bool | None],
summary="删除报价单及明细信息",
dependencies=[DependsJwtAuth],
)
async def delete_quotation_related(
id: Annotated[str | None, Path(description="报价单ID")],
request: Request,
db: AsyncSession = Depends(get_db),
):
# ip = await get_request_ip(request)
# 获取记录详情
item = await quotation_crud.get(db, id)
if not item:
return AjaxResponse(False)
# 移除相关的明细记录
res = await quotationdetail_crud.delete_by_column(db, "orderno", item.handno)
if res:
# 然后删除报价单
await quotation_crud.delete_byid(db, id)
return AjaxResponse(res)
使用基类控制器,可以继承常规CRUD的接口,并自动生成路由、依赖注入、数据库连接等功能 ——构建方式2
controller = BaseController[Quotation, str, QuotationPageDto, QuotationDto]( quotation_crud, pagedto_class=QuotationPageDto, dto_class=QuotationDto, router=router, )
controller.init_router() # 初始化常规CRUD等接口的路由
复制代码
而报价单明细业务控制器,主要需要根据报价单号(订单号)获取详情的接口
复制代码
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Path, Body
from typing import Type, TypeVar, Generic, List, Dict, Any, Optional, Annotated
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
from schemas.base import AjaxResponse, ErrorInfo, ListResult, PagedResult
from api.base_controller import BaseController
from db.session_async import get_db
from common.jwt import DependsJwtAuth
from models.quotationdetail import QuotationDetail
from schemas.quotationdetail import QuotationDetailDto, QuotationDetailPageDto
from crud.quotationdetail import quotationdetail as quotationdetail_crud
from crud.operationlog import operationlog as operationlog_crud
from utils.request_parse import get_request_ip
# 创建路由,用于处理自定义接口
router = APIRouter()
@router.get(
"/by-orderno",
response_model=AjaxResponse[List[QuotationDetailDto] | None],
summary="根据订单编号获取记录",
dependencies=[DependsJwtAuth],
)
async def get_by_orderno(
orderno: Annotated[str | None, Query(description="订单编号")],
request: Request,
db: AsyncSession = Depends(get_db),
):
# ip = await get_request_ip(request)
items = await quotationdetail_crud.get_by_orderno(db, orderno)
# print(len(items))
items = [QuotationDetailDto.model_validate(item) for item in items]
return AjaxResponse(items)
@router.delete(
"/by-orderno/{orderno}",
response_model=AjaxResponse[bool | None],
summary="根据订单编号删除所有详情记录",
dependencies=[DependsJwtAuth],
)
async def delete_by_orderno(
orderno: Annotated[str | None, Path(description="订单编号")],
request: Request,
db: AsyncSession = Depends(get_db),
):
res = await quotationdetail_crud.delete_by_attributes(db, QuotationDetail.orderno == orderno)
return AjaxResponse(res)
# 使用基类控制器,可以继承常规CRUD的接口,并自动生成路由,依赖注入,数据库连接等功能
controller = BaseController[QuotationDetail, str, QuotationDetailPageDto, QuotationDetailDto]( quotationdetail_crud, pagedto_class=QuotationDetailPageDto, dto_class=QuotationDetailDto, router=router, )
controller.init_router() # 初始化常规CRUD等接口的路由
由于基类控制器接口的标准化,我们根据框架后端的接口进行前端API调用类的封装处理,从而实现业务基类调用接口的一致性封装,简化代码。这样增删改查等操作的接口都可以抽象到BaseApi里面了。
例如权限模块中的用户管理、机构管理、角色管理、菜单管理、功能管理、操作日志和登录日志等业务类,这些类继承BaseApi后,将具备相关的接口,如下所示的继承关系。
2、基于SqlAlchemy实现的业务模型
FastAPI负责提供数据的API接口,底层的数据处理通过SqlAlchemy + MongoDB 实现多数据库的数据管理,如对接MySQL、Postgresql、SQLite、Oracle和MongoDB等进行数据交互处理。
对于常规业务表,我们采用SqlAlchemy实现数据库的ORM管理。SqlAlchemy是Python领域中一个非常强大的ORM管理工具之一, 它让你用 Python 对象来操作数据库,而不是手写 SQL 语句。
通常我们先定义好模型的基类,提供简单的封装
[此处为图片1]
然后在业务类里继承它即可
[此处为图片2]
报价单的明细表也是类似的
[此处为图片3]
这些模型类和对应接口的DTO类只需在代码生成工具中一键生成,无需手动编写。
使用代码生成工具 Database2Sharp 打开数据库列表后,右键菜单可以选择生成对应的Python+FastApi后端项目,如下界面所示。
选中相关的表后,一键可以生成各层的类文件,其中包括最复杂的Model映射类信息。以下是生成的相关类的界面效果。
[此处为图片4]
2、WxPython实现基于Python桌面端主从表的数据处理
我们知道,对于单表来说,业务和界面会相对简单,如下图所示的是在Windows下客户信息的列表管理和数据编辑界面。
而对于主从表,通常除了主业务表外,还会关联一个或多个明细表。以报价单为例,只有一个明细表,具体界面展示如下。
[此处为图片5]
对于列表的主从表关联关系,主要是增加了一个明细表的处理和展示。
在主从表编辑界面上,则需要进行更复杂的处理,在表格中直接录入并保存明细的操作。以下是主从表的编辑界面实现效果。
[此处为图片6]
主要是在首次创建时,对表格数据类进行设置
[此处为图片7]
表格的数据直接录入,通常不只是通过文本框录入,还包括选择表记录、下拉列表、复选框、图像、数值和颜色等特殊输入方式。
例如自定义数据列表选择界面,通过定义一个产品的数据列表供选择,单击产品编码处时,会弹出一个选择框进行选择。
[此处为图片8]
为了实现对表格数据单元格的点击监控,我们绑定了对应的事件。
AsyncBind(wx.grid.EVT_GRID_CELL_LEFT_CLICK, self.on_cell_left_click, self.sub_grid)
然后对事件进行处理即可。
[此处为图片9]
下拉列表则通过绑定固定列表或字典类型来实现字典下拉选择
[此处为图片10]
初始化字典列表非常简单,如下代码所示。
[此处为图片11]
其他示例可以参考测试效果,支持多种数据输入处理,以下是测试界面效果。
[此处为图片12]


雷达卡


京公网安备 11010802022788号







