使用SQLAlchemy实现三表嵌套JOIN查询获取第三表ID的方法问询

前提说明 #

首先需要你提前定义好三张表对应的ORM映射类，字段属性、关联关系和实际表结构一致即可，参考示例如下：

from sqlalchemy import Column, Integer, ForeignKey, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class TableOne(Base):
    __tablename__ = "table_one"
    id = Column(Integer, primary_key=True)
    # 其他字段按需补充

class TableTwo(Base):
    __tablename__ = "table_two"
    id = Column(Integer, primary_key=True)
    table_one_id = Column(Integer, ForeignKey("table_one.id"))
    external_id = Column(String) # 字段类型可根据实际情况调整

class TableThree(Base):
    __tablename__ = "table_three"
    id = Column(Integer, primary_key=True)
    table_two_id = Column(Integer, ForeignKey("table_two.id"))
    external_id = Column(String) # 字段类型可根据实际情况调整

SQLAlchemy 2.0 写法（当前主流推荐版本） #

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

# 替换为实际的数据库连接串
engine = create_engine("mysql+pymysql://user:password@host:port/db_name")

# 替换为实际查询参数
x = 1
y = "external_id_y_value"
z = "external_id_z_value"

with Session(engine) as session:
    stmt = select(TableThree.id)\
        .join(TableTwo, TableThree.table_two_id == TableTwo.id)\
        .join(TableOne, TableTwo.table_one_id == TableOne.id)\
        .where(
            TableOne.id == x,
            TableTwo.external_id == y,
            TableThree.external_id == z
        )
    # 取单个结果用scalar，取多条匹配结果用scalars().all()
    result = session.scalar(stmt)

SQLAlchemy 1.x 旧版本写法 #

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 替换为实际的数据库连接串
engine = create_engine("mysql+pymysql://user:password@host:port/db_name")
Session = sessionmaker(bind=engine)
session = Session()

# 替换为实际查询参数
x = 1
y = "external_id_y_value"
z = "external_id_z_value"

result = session.query(TableThree.id)\
    .join(TableTwo, TableThree.table_two_id == TableTwo.id)\
    .join(TableOne, TableTwo.table_one_id == TableOne.id)\
    .filter(
        TableOne.id == x,
        TableTwo.external_id == y,
        TableThree.external_id == z
    ).first()
# 最终取值
target_id = result[0] if result else None

补充说明 #

你给出的原生SQL嵌套JOIN逻辑和上述顺序JOIN写法完全等价，SQLAlchemy生成的执行SQL和需求一致，如果你需要校验生成的SQL是否正确，直接打印stmt对象（2.0版本）或者query对象（1.x版本）即可查看。

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