Sanic中SQLAlchemy Asyncio查询问题：过滤订单内卖家自有产品

我是Sanic和SQLAlchemy新手，正在开发C2C电商系统。订单与产品通过order_products表（含quantity额外字段）建立多对多关联。需求为提取状态为complete、包含当前卖家产品的订单，且返回产品及对应数量，但现有查询结果混入了其他卖家的产品。

相关模型定义 #

class Order(Base):
    __tablename__ = "orders"
    uuid = Column(String(), primary_key=True)
    userId = Column(String(), nullable=False)
    status = Column(String(), nullable=False, default="draft")

    products = relationship(
        "Product", secondary="order_products", back_populates='orders')

    order_products = relationship(
        "OrderProduct", back_populates="order")

    def to_dict(self):
        return {"uuid": self.uuid, "userId": self.userId, "status": self.status, "products": [{
            "title": product.title,
            "price": product.price,
            "description": product.description,
            "uuid": product.uuid,
            "userId": product.userId,
            "quantity": product.order_products.quantity
        } for product in self.products]}

class Product(Base):
    __tablename__ = "products"
    uuid = Column(String(), primary_key=True)
    title = Column(String(), nullable=False)
    price = Column(FLOAT(), nullable=False)
    description = Column(String(), nullable=False)
    userId = Column(String(), nullable=False)

    orders = relationship(
        "Order", secondary="order_products", back_populates='products', uselist=False)

    order_products = relationship(
        "OrderProduct", back_populates="product")

class OrderProduct(Base):
    __tablename__ = "order_products"
    id = Column(Integer, primary_key=True)
    orderId = Column(String, ForeignKey("orders.uuid"))
    productId = Column(String, ForeignKey("products.uuid"))
    quantity = Column(Integer)

    # Define the relationships to the Order and Product tables
    order = relationship("Order", back_populates="order_products")
    product = relationship("Product", back_populates="order_products")

期望输出 #

仅包含当前卖家产品的已完成订单：

[
    {
        "uuid": "order_Id_abc",
        "userId": "user_Id_def", // 下单用户ID
        "status": "complete",
        "products": [
            {
                "title": "One Piece t-shirt",
                "price": 900.0,
                "description": "Luffy at laugh tail island",
                "uuid": "product_Id",
                "userId": "current_user_Id", // 卖家ID
                "quantity": 2
            }
        ]
    }
]

现有查询语句 #

from sqlalchemy.orm import selectinload
from sqlalchemy.sql.expression import and_
from sqlalchemy.future import select

q = (select(Order)
         .join(OrderProduct, Order.uuid == OrderProduct.orderId)
         .join(Product, OrderProduct.productId == Product.uuid)
         .filter(and_((Order.status == 'complete'), (Product.userId == "current_user_id")))
         .options(selectinload(Order.products).selectinload(Product.order_products))
         .group_by(Order.uuid)
         )

问题分析 #

1. 关联加载逻辑错误：现有查询通过join过滤出了包含当前卖家产品的订单，但selectinload(Order.products)会无条件加载该订单下的所有关联产品（包括其他卖家的），这是导致结果混入非目标产品的核心原因。
2. Product模型关系配置错误：orders = relationship(..., uselist=False)不符合多对多关联的逻辑——一个产品可以被多个订单购买，应该去掉uselist=False，否则会导致关联数据加载异常。
3. quantity获取方式错误：product.order_products.quantity写法错误，因为一个Product对应多个OrderProduct实例（关联不同订单），直接访问会抛出"AttributeError: 'InstrumentedList' object has no attribute 'quantity'"，必须通过订单的order_products关联匹配对应的产品数量。

解决方案 #

1. 修正Product模型的关联配置 #

去掉orders关系中的uselist=False：

class Product(Base):
    __tablename__ = "products"
    # ... 其他字段不变
    orders = relationship(
        "Order", secondary="order_products", back_populates='products')
    # ... 其他关联不变

2. 修改查询语句：精准加载目标产品 #

使用contains_eager结合joinedload，确保只加载当前卖家的产品，替代原来的selectinload：

from sqlalchemy.orm import joinedload, contains_eager
from sqlalchemy.sql.expression import and_
from sqlalchemy.future import select

q = (select(Order)
     .join(OrderProduct, Order.uuid == OrderProduct.orderId)
     .join(Product, OrderProduct.productId == Product.uuid)
     .filter(and_(Order.status == 'complete', Product.userId == "current_user_id"))
     .options(
         joinedload(Order.order_products).joinedload(OrderProduct.product),
         contains_eager(Order.products).filter(Product.userId == "current_user_id")
     )
     .group_by(Order.uuid, Product.uuid, OrderProduct.id)
     )

3. 修正to_dict方法：正确获取产品数量 #

通过Order的order_products关联匹配对应产品的quantity：

class Order(Base):
    # ... 其他字段和关联不变
    def to_dict(self):
        # 先构建订单产品的映射：product_uuid -> quantity
        product_quantity_map = {op.product.uuid: op.quantity for op in self.order_products}
        return {
            "uuid": self.uuid,
            "userId": self.userId,
            "status": self.status,
            "products": [
                {
                    "title": product.title,
                    "price": product.price,
                    "description": product.description,
                    "uuid": product.uuid,
                    "userId": product.userId,
                    "quantity": product_quantity_map.get(product.uuid, 0)
                }
                for product in self.products
                # 额外过滤，确保只保留当前卖家的产品
                if product.userId == "current_user_id"
            ]
        }

内容的提问来源于stack exchange，提问作者Shreyas Chorge