如何在Python Peewee中实现子查询的自交叉连接

问题背景 #

已成功用Peewee还原SQL中的common_subquery公共子查询，但无法实现该子查询与自身的交叉连接（CROSS JOIN），要求在数据库引擎端完成计算而非Python内存中处理。

解决方案步骤 #

1. 定义基础模型别名与公共子查询

首先为Flights模型创建两个别名，用于构建公共子查询的自连接：

from peewee import *

# 为Flights模型创建两个别名，对应SQL中的t1和t2
t1 = Flights.alias()
t2 = Flights.alias()

# 构建common_subquery公共子查询
common_subquery = (
    t1.select(
        t1.fly_from,
        t1.airlines.alias('first_airline'),
        t1.flight_numbers.alias('first_flight_number'),
        t1.link_to.alias('first_link'),
        t1.departure_to,
        t1.fly_to.alias('connection_at'),
        t2.airlines.alias('second_airline'),
        t2.flight_numbers.alias('second_flight_number'),
        t2.link_to.alias('second_link'),
        t2.fly_to,
        t1.arrival_to.alias('landing_at_connection'),
        t2.departure_to.alias('departure_from_connection'),
        t2.arrival_to,
        # 计算中转时长（小时）
        fn.Cast(
            (fn.julianday(t2.departure_to) - fn.julianday(t1.arrival_to)) * 24,
            Integer
        ).alias('duration_hours'),
        # 计算两段航班总价
        (t1.discount_price + t2.discount_price).alias('total_price')
    )
    .join(t2, on=(t1.flight_hash == t2.flight_hash))
    .where(
        (t2.fly_from != t1.fly_from),
        (t1.fly_from != t2.fly_to)
    )
    .order_by(SQL('total_price ASC'))
    .cte('common_subquery')  # 将查询标记为CTE
)

2. 实现CTE自交叉连接

核心要点是为CTE创建两个独立的别名，这样Peewee才能区分SQL中的t1和t2实例，避免冲突：

# 为CTE创建两个别名，对应主查询中的t1和t2
t1_sub = common_subquery.alias()
t2_sub = common_subquery.alias()

# 构建最终的主查询
final_query = (
    t1_sub.select(
        # 选择t1_sub（去程中转航班）的字段
        t1_sub.fly_from.alias('source'),
        t1_sub.first_airline.alias('source_outbound_airline'),
        t1_sub.first_flight_number.alias('source_outbound_flight_number'),
        t1_sub.first_link.alias('source_outbound_link'),
        t1_sub.departure_to.alias('outbound_departure'),
        t1_sub.landing_at_connection,
        t1_sub.connection_at.alias('outbound_connection'),
        t1_sub.second_airline.alias('connection_outbound_airline'),
        t1_sub.second_flight_number.alias('connection_outbound_flight_number'),
        t1_sub.second_link.alias('connection_outbound_link'),
        t1_sub.departure_from_connection,
        t1_sub.arrival_to.alias('destination_arrival'),
        t1_sub.fly_to.alias('destination'),
        
        # 选择t2_sub（返程中转航班）的字段
        t2_sub.first_airline.alias('inbound_connection_airline'),
        t2_sub.first_flight_number.alias('inbound_connection_flight_number'),
        t2_sub.first_link.alias('inbound_connection_link'),
        t2_sub.departure_to.alias('return_departure'),
        t2_sub.landing_at_connection.alias('return_arrival'),
        t2_sub.connection_at.alias('inbound_connection'),
        t2_sub.second_airline.alias('inbound_airline'),
        t2_sub.second_flight_number.alias('inbound_flight_number'),
        t2_sub.second_link.alias('inbound_link'),
        t2_sub.departure_from_connection.alias('return_departure_from'),
        t2_sub.arrival_to.alias('return_destination_arrival'),
        
        # 计算总价和停留天数
        (fn.Ceil((t1_sub.total_price + t2_sub.total_price) / 100.0) * 100).alias('round_total_price'),
        fn.Floor(
            fn.julianday(t2_sub.departure_from_connection) - fn.julianday(t1_sub.arrival_to)
        ).alias('days_in_dest')
    )
    # 执行CTE自交叉连接
    .select_from(t1_sub.cross_join(t2_sub))
    .where(
        # 停留天数限制在5-8天
        (fn.julianday(t2_sub.departure_from_connection) - fn.julianday(t1_sub.arrival_to)).between(5, 8),
        # 中转时长不超过24小时
        t1_sub.duration_hours < 24,
        t2_sub.duration_hours < 24,
        # 去程终点等于返程起点
        t1_sub.fly_to == t2_sub.fly_from,
        # 筛选出发地含TLV、目的地含PRG的航班
        t1_sub.fly_from.contains('TLV'),
        t1_sub.fly_to.contains('PRG')
    )
    .order_by(
        SQL('round_total_price ASC'),
        t1_sub.duration_hours.asc(),
        t2_sub.duration_hours.asc()
    )
    .with_cte(common_subquery)  # 将CTE关联到主查询
)

3. 关键说明

• 别名区分：必须为CTE创建两个独立别名（t1_sub和t2_sub），不能直接复用同一个CTE对象，否则Peewee无法生成正确的SQL别名。
• 交叉连接实现：使用cross_join()方法等价于SQL中的CROSS JOIN，也可以替换为.join(t2_sub, JOIN.CROSS)，效果一致。
• 数据库函数调用：通过fn对象调用SQL内置函数（如julianday、Ceil），Peewee会自动适配目标数据库的语法。
• 模糊匹配：用contains()方法实现SQL中的LIKE '%xxx%'逻辑。

内容的提问来源于stack exchange，提问作者Meir Tolpin