如何用SQLAlchemy实现带子查询ON条件的LEFT JOIN？

背景 #

现有SQLAlchemy实体类EmployeePayRate和User，需要实现左连接带子查询ON条件的逻辑，获取指定日期前员工最近的有效薪资记录。对应目标MySQL查询已给出，但现有SQLAlchemy代码触发以下错误：

InvalidRequestError("Select statement '<sqlalchemy.sql.selectable.Select object at 0x0000024EB4C4B320>' returned no FROM clauses due to auto-correlation; specify correlate() to control correlation manually.")

实体类代码 #

class EmployeePayRate(Base):
    __tablename__ = "employee_pay_rates"

    pay_rate_id:Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    user_id:Mapped[int] = mapped_column(ForeignKey(User.user_id))
    company_id: Mapped[int] = mapped_column(ForeignKey(Company.company_id))
    pay_rate: Mapped[float]
    charge_rate: Mapped[float]
    active: Mapped[bool] = mapped_column(default = True)
    deleted: Mapped[bool] = mapped_column(default = False)
    created_date: Mapped[datetime.datetime] = mapped_column(DateTime(timezone=True), server_default = text('CURRENT_TIMESTAMP'))
    start_date: Mapped[datetime.datetime] = mapped_column(DateTime(timezone=True))


class User(Base):
    __tablename__ = "users"

    user_id:Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    company_id: Mapped[int] = mapped_column(ForeignKey(Company.company_id))
    full_name: Mapped[str] = mapped_column(String(75), default ='')
    email: Mapped[str] = mapped_column(String(255), default ='')
    phone: Mapped[str] = mapped_column(String(25), default ='')
    lang: Mapped[str] = mapped_column(String(25), default ='')
    time_zone: Mapped[str] = mapped_column(String(50), default ='')

目标MySQL查询 #

SELECT u.user_id, u.full_name, epr.start_date
FROM users as u
LEFT JOIN employee_pay_rates as epr on epr.pay_rate_id =  (select epr1.pay_rate_id
            from employee_pay_rates as epr1
            WHERE epr1.start_date <= '2024-08-24'
            AND epr1.company_id = u.company_id AND epr1.user_id = u.user_id
            ORDER BY epr1.start_date LIMIT 1)
WHERE u.company_id = 1

报错的现有SQLAlchemy代码 #

payrate_sel_stmt = select (EmployeePayRate).where(
    and_(
        EmployeePayRate.company_id == User.company_id,
        EmployeePayRate.user_id == User.user_id, 
        cast(EmployeePayRate.start_date, Date) >= datetime.datetime.now().date
    )
).order_by(EmployeePayRate.start_date.desc()).limit(1)

test_user_sel_stmt = select(User).outerjoin(EmployeePayRate, EmployeePayRate.pay_rate_id == payrate_sel_stmt).where(
    User.company_id == data["company_id"]
)
users = session.execute(test_user_sel_stmt)

错误原因 #

1. 子查询直接引用外部User表但未明确关联规则，导致SQLAlchemy无法自动处理关联关系，触发自动关联错误。
2. 子查询错误地选择整个EmployeePayRate实体，而非目标字段pay_rate_id。
3. 日期条件逻辑写反（目标需求是<= 指定日期，原代码写为>=）。

修正后的SQLAlchemy代码 #

from sqlalchemy import select, and_, cast, Date, text
import datetime

# 定义目标查询日期，替换为业务实际需要的日期
target_date = datetime.date(2024, 8, 24)

# 子查询：获取每个用户在目标日期前最近的有效薪资记录ID
payrate_sel_stmt = select(EmployeePayRate.pay_rate_id).where(
    and_(
        EmployeePayRate.company_id == User.company_id,
        EmployeePayRate.user_id == User.user_id,
        cast(EmployeePayRate.start_date, Date) <= target_date,
        EmployeePayRate.active == True,  # 可选：过滤有效记录
        EmployeePayRate.deleted == False
    )
).order_by(EmployeePayRate.start_date.desc()).limit(1).correlate(User)  # 明确关联外部User表

# 主查询：左连接子查询结果，返回指定字段
test_user_sel_stmt = select(
    User.user_id,
    User.full_name,
    EmployeePayRate.start_date,
    EmployeePayRate.pay_rate,
    EmployeePayRate.charge_rate
).outerjoin(
    EmployeePayRate,
    EmployeePayRate.pay_rate_id == payrate_sel_stmt
).where(
    User.company_id == data["company_id"]
)

# 执行查询并获取结果
result = session.execute(test_user_sel_stmt).all()

关键修正点 #

• 子查询明确选择pay_rate_id字段，匹配目标SQL的逻辑。
• 添加.correlate(User)指定子查询关联外部的User表，解决自动关联报错。
• 修正日期条件为<= target_date，符合“指定日期前最近记录”的需求。
• 可选添加active和deleted条件，确保只查询有效记录。
• 主查询明确指定返回字段，避免返回冗余的实体对象。

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