qidao123.com技术社区-IT企服评测·应用市场

标题: FastAPI中的复杂查询与原子更新指南 [打印本页]
作者: 拉不拉稀肚拉稀    时间: 2025-5-2 20:52
标题: FastAPI中的复杂查询与原子更新指南
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:
tags:
扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
探索数千个预构建的 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的条件表达式构造器，支持逻辑运算符：
示例：查询金额大于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

余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长，阅读完整的文章：FastAPI中的复杂查询与原子更新指南 | cmdragon's Blog
往期文章归档：


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



欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/) Powered by Discuz! X3.4