发帖
雷达卡
栈挥鹊鞍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）处理函数。
[此处为图片2]我们来看看针对产品报价单和明细记录的处理。这两个代表不同的业务表，可以通过管理它们的记录关系实现主从表的灵活性。
对于主表，在基础接口之外定义两个额外的函数，主要是在删除主表的同时执行明细记录的关联操作。
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接口，并自动生成路由、依赖注入和数据库连接等功能——构建方式2controller = 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),
):
items = await quotationdetail_crud.get_by_orderno(db, orderno)
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、WxPthon实现基于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号
