如何使用Peewee Python ORM对交叉连接子查询进行预取？

首先我先补全你未写完的Actual模型（假设结构和Plan类似，用于存储实际销量数据），然后分两种常用场景来实现你的需求：

import peewee

class BaseModel(peewee.Model):
    class Meta:
        database = peewee.SqliteDatabase(':memory:')  # 替换为你的实际数据库配置

class Product(BaseModel):
    product_id = peewee.CharField(primary_key=True)
    list_price = peewee.FloatField()

class Week(BaseModel):
    week_date = peewee.DateField(primary_key=True)

class Plan(BaseModel):
    product = peewee.ForeignKeyField(Product, backref='plans')
    week = peewee.ForeignKeyField(Week, backref='plans')
    sales = peewee.FloatField(null=True)
    class Meta:
        primary_key = peewee.CompositeKey('product', 'week')

class Actual(BaseModel):
    product = peewee.ForeignKeyField(Product, backref='actuals')
    week = peewee.ForeignKeyField(Week, backref='actuals')
    actual_sales = peewee.FloatField(null=True)
    class Meta:
        primary_key = peewee.CompositeKey('product', 'week')

# 初始化表（仅首次运行需要）
BaseModel.create_tables([Product, Week, Plan, Actual])

方案1：SQL层面直接实现交叉连接+左关联（大数据量推荐） #

这种方式通过一次SQL查询完成笛卡尔积生成和关联数据匹配，效率更高，适合数据量较大的场景：

# 构建交叉连接（Product × Week）+左关联Plan和Actual的查询
query = (Product
         .select(
             Product.product_id,
             Week.week_date,
             Plan.sales.alias('plan_sales'),
             Actual.actual_sales.alias('actual_sales')
         )
         .join(Week, on=None)  # on=None 触发SQL的CROSS JOIN
         .join(Plan, on=(
             (Product.product_id == Plan.product) & (Week.week_date == Plan.week)
         ), join_type=peewee.JOIN.LEFT_OUTER)
         .join(Actual, on=(
             (Product.product_id == Actual.product) & (Week.week_date == Actual.week)
         ), join_type=peewee.JOIN.LEFT_OUTER))

# 遍历查询结果
for row in query.tuples():
    product_id, week_date, plan_sales, actual_sales = row
    print(f"Product: {product_id}, Week: {week_date}")
    print(f"  计划销量: {plan_sales if plan_sales else '无数据'}")
    print(f"  实际销量: {actual_sales if actual_sales else '无数据'}")

方案2：内存中生成交叉组合并匹配关联数据（小数据量推荐） #

如果数据量不大，这种方式代码更直观，不需要复杂的SQL拼接：

# 先批量加载所有基础数据到内存
products = {p.product_id: p for p in Product.select()}
weeks = {w.week_date: w for w in Week.select()}
plans = {(p.product_id, p.week.week_date): p for p in Plan.select()}
actuals = {(a.product_id, a.week.week_date): a for a in Actual.select()}

# 生成所有Product×Week的交叉组合，并匹配对应的Plan和Actual
class ProductWeekCombo:
    def __init__(self, product, week, plan=None, actual=None):
        self.product = product
        self.week = week
        self.plan = plan
        self.actual = actual

combinations = []
for product in products.values():
    for week in weeks.values():
        key = (product.product_id, week.week_date)
        combinations.append(ProductWeekCombo(
            product=product,
            week=week,
            plan=plans.get(key),
            actual=actuals.get(key)
        ))

# 遍历组合数据
for combo in combinations:
    print(f"Product: {combo.product.product_id}, Week: {combo.week.week_date}")
    print(f"  计划销量: {combo.plan.sales if combo.plan else '无数据'}")
    print(f"  实际销量: {combo.actual.actual_sales if combo.actual else '无数据'}")

注意事项 #

Peewee默认的prefetch方法是基于外键的关联查询，适合常规的一对多/多对多场景，但交叉连接属于笛卡尔积，没有直接的外键关联关系，因此需要通过上述两种方式实现需求。

内容的提问来源于stack exchange，提问作者Mike Drawback