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. 环境准备与模型定义

在开始前确保已安装必要依赖：

pip install fastapi uvicorn tortoise-orm pydantic

创建订单模型示例（models.py）：

from tortoise.models import Model
from tortoise import fields


class Product(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255)
    stock = fields.IntField(default=0)
    price = fields.DecimalField(max_digits=10, decimal_places=2)


class Order(Model):
    id = fields.IntField(pk=True)
    status = fields.CharField(max_length=20)  # pending/completed/canceled
    total_amount = fields.DecimalField(max_digits=10, decimal_places=2)
    product = fields.ForeignKeyField('models.Product', related_name='orders')
    created_at = fields.DatetimeField(auto_now_add=True)

创建对应的Pydantic模型（schemas.py）：

from pydantic import BaseModel
from datetime import datetime


class OrderOut(BaseModel):
    id: int
    status: str
    total_amount: float
    product_id: int
    created_at: datetime

    class Config:
        orm_mode = True

2. 组合Q对象实现复杂查询

2.1 Q对象基础原理

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

& 表示AND
| 表示OR
~ 表示NOT

示例：查询金额大于100且状态为pending的订单

from tortoise.expressions import Q


async def get_orders():
    return await Order.filter(
        Q(total_amount__gt=100) & Q(status="pending")
    ).all()

2.2 多条件动态组合

在路由中实现动态过滤（main.py）：

from fastapi import APIRouter, Query
from tortoise.expressions import Q

router = APIRouter()


@router.get("/orders", response_model=list[OrderOut])
async def search_orders(
        min_amount: float = Query(None),
        max_amount: float = Query(None),
        status: str = Query(None)
):
    query = Q()
    if min_amount:
        query &= Q(total_amount__gte=min_amount)
    if max_amount:
        query &= Q(total_amount__lte=max_amount)
    if status:
        query &= Q(status=status)

    return await Order.filter(query).prefetch_related('product')

2.3 复杂逻辑示例

查询过去7天内金额超过500的已完成订单，或金额低于100的待处理订单：

from datetime import datetime, timedelta


async def complex_query():
    seven_days_ago = datetime.now() - timedelta(days=7)
    return await Order.filter(
        Q(
            Q(created_at__gte=seven_days_ago) &
            Q(total_amount__gt=500) &
            Q(status='completed')
        ) |
        Q(
            Q(total_amount__lt=100) &
            Q(status='pending')
        )
    ).order_by('-created_at')

3. 使用F表达式进行原子更新

3.1 F表达式的作用原理

F表达式直接在数据库层面执行运算，避免竞态条件。示例：安全扣减库存

from tortoise.expressions import F


async def decrease_stock(product_id: int, quantity: int):
    await Product.filter(id=product_id).update(
        stock=F('stock') - quantity
    )

3.2 复合更新操作

同时更新多个字段：

async def update_product_price(product_id: int, new_price: float):
    await Product.filter(id=product_id).update(
        price=new_price,
        last_updated=datetime.now(),
        version=F('version') + 1
    )

3.3 条件更新示例

只有当库存充足时才允许扣减：

async def safe_purchase(product_id: int, quantity: int):
    updated = await Product.filter(
        id=product_id,
        stock__gte=quantity
    ).update(stock=F('stock') - quantity)

    if not updated:
        raise HTTPException(400, "库存不足")

4. 完整案例演示

实现商品促销接口：

@router.post("/products/{product_id}/promotion")
async def create_promotion(
        product_id: int,
        discount: float = Body(..., gt=0, lt=1)
):
    # 原子更新价格并记录操作
    updated = await Product.filter(id=product_id).update(
        price=F('price') * (1 - discount),
        promotion_count=F('promotion_count') + 1
    )

    if not updated:
        raise HTTPException(404, "商品不存在")

    # 获取更新后的对象
    product = await Product.get(id=product_id)
    return {
        "new_price": float(product.price),
        "promotion_count": product.promotion_count
    }

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

为什么要使用F表达式而不是先查询再更新？
A) 减少数据库查询次数
B) 避免并发导致的数据不一致
C) 两种方式效果相同
答案：B。F表达式保证原子操作，防止竞态条件
如何防止通过Q对象构造的查询出现SQL注入？
A) 手动转义参数
B) 使用ORM的内置参数化查询
C) 限制用户输入字段
答案：B。Tortoise-ORM会自动处理查询参数化

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

往期文章归档：

ASCII编码的影响与作用：数字化时代的不可或缺之物

二月 25, 2024

一、ASCII编码的起源 ASCII（American Standard Code for Information Interchange）编码是一种最早用于将字符转换为数字的编码系统。它诞生于20世纪60年代，旨在解决计算机系统之间的字符传输和存储问题。在ASCII编码出现之前，不同的计算机系统使用不同的字符编码，导致字符在系统之间的传输和处理变得困难。 Ascii编码解码 | 一个覆盖广泛主题工具的高效在线平台(amd794.com) https://amd794.com/asciiencordec 二、ASCII编码解决了什么问题 ASCII编码的出现解决了字符传输和存储的标准化问题。它将字符映射为固定的7位二进制数，共计128个字符，包括英文字母、数字、标点符号和控制字符等。这种标准化的编码方式使得不同计算机系统之间可以准确地传输和处理字符数据，大大简化了字符处理的复杂性。三、ASCII编码对现在的影响和作用文本处理：ASCII编码在文本处理中起着重要作用，包括文件存储、数据传输、网络通信等。它成为了计算机系统中最基本的字符表示方式，被广泛应用于各种文本处理场景。编程语言：ASCII编码是大多数编程语言中字符表示的基础。程序员可以使用ASCII编码来处理和操作文本数据，实现各种字符处理的功能。数据库存储：ASCII编码可以用于将文本数据存储在数据库中，方便数据的检索和处理。它提供了一种标准的字符表示方式，使得数据的存储和查询更加方便和高效。跨平台兼容性：ASCII编码是一种标准化的字符编码系统，可以在不同的计算机和操作系统中使用。这种跨平台兼容性使得文本数据能够在不同的环境中无缝传输和处理。多语言处理：尽管ASCII编码只能表示英文字母、数字和一些基本符号，但它为后续的字符编码系统奠定了基础。ASCII编码的发展促使了后续的字符编码系统的出现，满足了多语言文本处理的需求。四、ASCII编码的应用领域电信通信：ASCII编码在电报和传真等通信方式中起着关键作用，实现了字符数据的传输和处理。网络通信：ASCII编码是互联网通信的基础，HTTP协议、电子邮件和网页等都使用ASCII编码来传输和处理文本数据。数据存储与处理：ASCII编码可以用于将文本数据存储在文件和数据库中，方便后续的数据检索和处理。编程语言与脚本：ASCII编码是大多数编程...

阅读全文

搜索此博客

cmdragon's Blog

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