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 后端的基类控制器定义。

 有了上面基类定义好的常规化接口，子类只需要继承基类控制器即可获得强大的功能接口了。

一般API控制器的子类，只需要继承基类就可以了，额外增加的接口按常规化的设计函数即可，可以参考基类的做法来写各种（GET、PUT、DELETE、POST）的处理函数。

我们来看看对于产品报价单和明细记录的处理，这两个代表不同的业务表，我们可以分开管理，通过约束他们的记录关系实现主从表弹性化的管理。

如对于主表，我们在基类接口外定义多两个函数，主要就是删除主表的时候，同时移除明细记录的关联操作。

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等接口的路由