如何在fastapi sqlalchemy postgresql中过滤关系？

Question

我有三个表：产品、last_solds 和客户。 last_solds 表保存销售给客户的产品的最后价格。我想要实现的是查询指定客户的last_sold的所有产品。我怎样才能做到这一点？

这是我尝试过的，但没有成功，结果仍然是所有产品都有last_solds。

课程：

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String, nullable=False)
    last_sold = relationship("LastProductSold", back_populates="product")

class Customer(Base):
    __tablename__ = "customers"
    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String, nullable=False)
    last_sold = relationship("LastProductSold", back_populates="customer")

class LastProductSold(Base):
    __tablename__ = "last_product_solds"
    id = Column(Integer, primary_key=True, nullable=False)
    price = Column(Numeric(precision=15, scale=2), nullable=False, default=0)
    product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
    customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False)
    product = relationship("Product", back_populates="last_sold")
    customer = relationship("Customer", back_populates="last_sold")

架构：

class LastProductSold(BaseModel):
    id: int
    price: float
    class Config:
        from_attributes = True

class Product(BaseModel):
    id: int
    name: str
    last_sold: Optional[list[LastProductSold]] = Field(default=None)
    class Config:
        from_attributes = True

端点：

@router.get("/products/customer/{customer_id}", response_model=list[Product])
def get_products_for_customer(
    customer_id: int,
    db: Session = Depends(get_db),
):
    products = db.query(Product)
                    .outerjoin(LastSold, LastSold.customer_id == customer_id)
                    .distinct()
    return products 

我也尝试过使用joinedload，但还是不行。如果可能的话我还需要了解 orm 是如何工作的，我认为我的理解仍然有点偏差。我尝试通过打印查询结果来进行调试，但它返回模型而不是对象，如果我不在端点上提供响应模型，那么当我访问路由时它会抛出错误。另一件事是，如果我尝试使用 .dict 方法将模型解析为 dict 来打印模型，我将看不到我的外连接结果。

我尝试在网上查找，但似乎找不到任何有关我的问题的相关解释，请帮助我。

Answer 1

您可以按照以下方式进行

from fastapi import Depends
from sqlalchemy import and_
from sqlalchemy.orm import Session
from typing import List

@router.get("/products/customer/{customer_id}", response_model=List[Product])
def get_products_for_customer(
    customer_id:int,
    db:Session = Depends(get_db),
):
    products_with_last_sold = (
        db.query(
            Product,
           LastProductSold.id.label("last_sold_id"),
            LastProductSold.price.label("last_sold_price"),
        )
        .outerjoin(
            LastProductSold,
            and_(
                LastProductSold.product_id == Product.id,
                LastProductSold.customer_id== customer_id,
            ),
        )
        .all()
    )
    products_list = []
    for product, last_sold_id, last_sold_price in products_with_last_sold:
        product_data = {
            'id': product.id,
            'name': product.name,
            'last_sold': None,
        }
        if last_sold_id is not None:
            product_data['last_sold'] = {
                'id':last_sold_id,
                'price': float(last_sold_price),
            }
        products_list.append(product_data)

    return products_list

因此，outerjoin 现在包含 customer_id 过滤器，确保仅指定客户的 last_sold 条目与每个产品连接。

并且我没有依赖last_sold关系，而是根据连接手动为每个产品构建了last_sold数据。