FastAPI中的复杂查询与原子更新指南

拉不拉稀肚拉稀  论坛元老 | 2025-5-2 20:52:14 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1798|帖子 1798|积分 5396

title: FastAPI中的复杂查询与原子更新指南
date: 2025/05/02 20:33:32
updated: 2025/05/02 20:33:32
author: cmdragon
excerpt:
FastAPI 结合 Tortoise-ORM 实现复杂查询与原子更新。通过 Q 对象构建多条件查询，支持 AND、OR、NOT 逻辑运算符，动态组合查询条件。使用 F 表达式进行原子更新，制止竞态条件，确保数据一致性。示例包罗订单状态与金额的复杂查询、库存扣减的原子操作，以及商品促销的价格更新。常见错误包罗字段拼写错误、类型不匹配和空结果集，需通过模型检查和异常处明白决。
categories:

  • 后端开发
  • FastAPI
tags:

  • FastAPI
  • Tortoise-ORM
  • 复杂查询
  • 原子更新
  • Q对象
  • F表达式
  • 数据库操作
扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
探索数千个预构建的 AI 应用，开启你的下一个巨大创意：https://tools.cmdragon.cn/
第一章：FastAPI复杂查询与原子更新实战

1. 环境准备与模型界说

在开始前确保已安装须要依赖：
  1. pip install fastapi uvicorn tortoise-orm pydantic
复制代码
创建订单模型示例（models.py）：
  1. from tortoise.models import Model
  2. from tortoise import fields
  5. class Product(Model):
  6.     id = fields.IntField(pk=True)
  7.     name = fields.CharField(max_length=255)
  8.     stock = fields.IntField(default=0)
  9.     price = fields.DecimalField(max_digits=10, decimal_places=2)
  12. class Order(Model):
  13.     id = fields.IntField(pk=True)
  14.     status = fields.CharField(max_length=20)  # pending/completed/canceled
  15.     total_amount = fields.DecimalField(max_digits=10, decimal_places=2)
  16.     product = fields.ForeignKeyField('models.Product', related_name='orders')
  17.     created_at = fields.DatetimeField(auto_now_add=True)
复制代码
创建对应的Pydantic模型（schemas.py）：
  1. from pydantic import BaseModel
  2. from datetime import datetime
  5. class OrderOut(BaseModel):
  6.     id: int
  7.     status: str
  8.     total_amount: float
  9.     product_id: int
  10.     created_at: datetime
  12.     class Config:
  13.         orm_mode = True
复制代码
2. 组合Q对象实现复杂查询

2.1 Q对象基础原理

Q对象是Tortoise-ORM的条件表达式构造器，支持逻辑运算符：

  • & 表示AND
  • | 表示OR
  • ~ 表示NOT
示例：查询金额大于100且状态为pending的订单
  1. from tortoise.expressions import Q
  4. async def get_orders():
  5.     return await Order.filter(
  6.         Q(total_amount__gt=100) & Q(status="pending")
  7.     ).all()
复制代码
2.2 多条件动态组合

在路由中实现动态过滤（main.py）：
  1. from fastapi import APIRouter, Query
  2. from tortoise.expressions import Q
  4. router = APIRouter()
  7. @router.get("/orders", response_model=list[OrderOut])
  8. async def search_orders(
  9.         min_amount: float = Query(None),
  10.         max_amount: float = Query(None),
  11.         status: str = Query(None)
  12. ):
  13.     query = Q()
  14.     if min_amount:
  15.         query &= Q(total_amount__gte=min_amount)
  16.     if max_amount:
  17.         query &= Q(total_amount__lte=max_amount)
  18.     if status:
  19.         query &= Q(status=status)
  21.     return await Order.filter(query).prefetch_related('product')
复制代码
2.3 复杂逻辑示例

查询过去7天内金额超过500的已完成订单，或金额低于100的待处理订单：
  1. from datetime import datetime, timedelta
  4. async def complex_query():
  5.     seven_days_ago = datetime.now() - timedelta(days=7)
  6.     return await Order.filter(
  7.         Q(
  8.             Q(created_at__gte=seven_days_ago) &
  9.             Q(total_amount__gt=500) &
  10.             Q(status='completed')
  11.         ) |
  12.         Q(
  13.             Q(total_amount__lt=100) &
  14.             Q(status='pending')
  15.         )
  16.     ).order_by('-created_at')
复制代码
3. 使用F表达式进行原子更新

3.1 F表达式的作用原理

F表达式直接在数据库层面实行运算，制止竞态条件。示例：安全扣减库存
  1. from tortoise.expressions import F
  4. async def decrease_stock(product_id: int, quantity: int):
  5.     await Product.filter(id=product_id).update(
  6.         stock=F('stock') - quantity
  7.     )
复制代码
3.2 复合更新操作

同时更新多个字段：
  1. async def update_product_price(product_id: int, new_price: float):
  2.     await Product.filter(id=product_id).update(
  3.         price=new_price,
  4.         last_updated=datetime.now(),
  5.         version=F('version') + 1
  6.     )
复制代码
3.3 条件更新示例

只有当库存充足时才允许扣减：
  1. async def safe_purchase(product_id: int, quantity: int):
  2.     updated = await Product.filter(
  3.         id=product_id,
  4.         stock__gte=quantity
  5.     ).update(stock=F('stock') - quantity)
  7.     if not updated:
  8.         raise HTTPException(400, "库存不足")
复制代码
4. 完整案例演示

实现商品促销接口：
  1. @router.post("/products/{product_id}/promotion")
  2. async def create_promotion(
  3.         product_id: int,
  4.         discount: float = Body(..., gt=0, lt=1)
  5. ):
  6.     # 原子更新价格并记录操作
  7.     updated = await Product.filter(id=product_id).update(
  8.         price=F('price') * (1 - discount),
  9.         promotion_count=F('promotion_count') + 1
  10.     )
  12.     if not updated:
  13.         raise HTTPException(404, "商品不存在")
  15.     # 获取更新后的对象
  16.     product = await Product.get(id=product_id)
  17.     return {
  18.         "new_price": float(product.price),
  19.         "promotion_count": product.promotion_count
  20.     }
复制代码
5. 常见报错解决方案

错误1：字段不存在

FieldError: Unknown field 'total_amout' for model Order
原因：字段名拼写错误（amout → amount）
解决：检查模型界说和查询字段是否一致
错误2：类型不匹配

ValidationError: 1 validation error for OrderOut...
原因：Decimal字段自动转换为float时精度丢失
解决：在Pydantic模型中使用Decimal类型并设置json_encoders
错误3：空结果集

DoesNotExist: Object does not exist
原因：查询条件过于严格导致无结果
解决：添加异常处理或使用first()取代get()
6. 课后Quiz


  • 当须要同时满足三个条件时，Q对象应该怎样组合？
    A) Q(a) | Q(b) | Q(c)
    B) Q(a) & Q(b) & Q(c)
    C) Q(a) & (Q(b) | Q(c))
    答案：B。& 运算符用于AND条件组合
  • 为什么要使用F表达式而不是先查询再更新？
    A) 淘汰数据库查询次数
    B) 制止并发导致的数据不一致
    C) 两种方式效果雷同
    答案：B。F表达式包管原子操作，防止竞态条件
  • 怎样防止通过Q对象构造的查询出现SQL注入？
    A) 手动转义参数
    B) 使用ORM的内置参数化查询
    C) 限定用户输入字段
    答案：B。Tortoise-ORM会自动处理查询参数化
余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长，阅读完整的文章：FastAPI中的复杂查询与原子更新指南 | cmdragon's Blog
往期文章归档：


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

拉不拉稀肚拉稀

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

渠道 国产数据库 集成商 AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表