如何使用具有子查询的 ON 语句进行 SQLAlchmey 左连接

Question

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 ='')

EmployeePayRate 可以有多个条目，这意味着某人的收费率或工资率可能会随着时间的推移而发生变化，当发生变化时，它会选择最新但早于给定日期的条目。 因此，如果我将 8-24-2024 作为下面数据中的日期要求，它将选择第二个。

员工工资率

67,37,1,2024-07-09 11:07:09,75.04,250.00,true,false,2024-07-09 11:07:09
73,37,1,2024-08-20 20:59:17,100.04,250.00,true,false,2024-08-20 20:59:17
75,37,1,2024-10-08 13:23:33,100.04,350.00,true,false,2024-10-08 13:23:33

用户

37,1,[email protected],1-ALAW-Z-1111222dd,[phone],ENG,EST

收到此错误：

InvalidRequestError(“由于自动关联，Select 语句 '' 没有返回 FROM 子句；指定 correlate() 来手动控制关联。”)

            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)

这是有效的 mysql 查询，我正在尝试在 sqlalchemy 中复制

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

Answer 1

在 SQLAlchemy 中，要使用子查询复制

LEFT JOIN

，尤其是涉及关联的子查询，您需要确保子查询正确关联以避免自动关联错误。以下是构建查询的方法。

问题分解：

1. 相关子查询：我们需要为每个用户获取最新的

   pay_rate_id

   ，其中

   start_date

   在特定日期之前。
2. LEFT JOIN：我们要使用此子查询的结果连接

   EmployeePayRate

   表。

解决方案：

我们可以通过在

pay_rate_id

的

ON

条件中使用

LEFT JOIN

的子查询来实现此目的。以下是在 SQLAlchemy 中编写查询的方法：

from sqlalchemy import select, and_
from sqlalchemy.orm import aliased

# Alias for the employee pay rates table
EmployeePayRateAlias = aliased(EmployeePayRate)

# Subquery to get the latest pay_rate_id before a specific date
subquery = (
    select(EmployeePayRateAlias.pay_rate_id)
    .where(
        and_(
            EmployeePayRateAlias.company_id == User.company_id,
            EmployeePayRateAlias.user_id == User.user_id,
            EmployeePayRateAlias.start_date <= '2024-08-24'  # Date parameter
        )
    )
    .order_by(EmployeePayRateAlias.start_date.desc())
    .limit(1)
    .correlate(User)  # Correlating the subquery with User table
)

# Main query to get the users and their most recent pay rates
test_user_sel_stmt = (
    select(User, EmployeePayRate)
    .outerjoin(EmployeePayRate, EmployeePayRate.pay_rate_id == subquery)
    .where(User.company_id == 1)  # Example company_id filter
)

# Execute the query
users = session.execute(test_user_sel_stmt)

说明：

1. 子查询:

   • 子查询从

     pay_rate_id

     表中选择

     EmployeePayRate

     ，其中

     start_date

     小于或等于

     '2024-08-24'

     。它通过

     User

     和

     company_id

     与

     user_id

     表相关。

     correlate(User)

     方法确保子查询链接到

     User

     表上的外部查询。

2. 主要查询：

   • 主查询从

     User

     表中进行选择，并使用子查询的结果对

     LEFT JOIN

     执行

     EmployeePayRate

     。这可确保您根据指定条件获得每个用户的最新付费率。

3. 关联子查询:

   • correlate(User)

     方法对于使子查询了解外部查询的

     User

     表至关重要，从而防止您遇到的“自相关”错误。

为什么有效：

这里的关键部分是显式使用

correlate()

来告诉 SQLAlchemy 子查询与

User

表上的外部查询相关。这样可以避免自相关错误并确保查询返回正确的结果。

通过遵循这种方法，您可以在 SQLAlchemy 中复制 MySQL 查询的行为，同时正确处理相关子查询。

---

参考资料：

• SQLAlchemy 文档：相关子查询
• SQLAlchemy 文档：子查询

如果您需要进一步说明，请告诉我！