SQLAlchemy关联子查询实现求助：基于给定ORM模型与SQL语句

给定的ORM模型 #

class Person(Base):
    __tablename__ = "stanovnik"

    id = Column(Integer, primary_key=True)
    ime = Column(String(20))
    prezime = Column(String(50))
    nadimak=Column(String(50),nullable=True)
    spol = Column(String(5))
    ratni_staz = Column(Integer)
    godine = Column(Integer)
    broj_clanova=Column(Integer,nullable=True)
    sifra_adr = Column(Integer,ForeignKey('adresa.sifra_adrese'))
    sifra_part = Column(Integer,ForeignKey('stanovnik.id'),nullable=True)
    sifra_ost = Column(Integer,ForeignKey('ostali.sifra_ostali'),nullable=True)
    adrese=relationship('Address')

class Address(Base):
    __tablename__ = "adresa"

    sifra_adrese=Column(Integer(),primary_key=True)
    naziv_adrese=Column(String(25))
    sifra_mje=Column(Integer(),ForeignKey('mjesto.sifra_mjesta'))
    mjesto=relationship('Village')

需求说明 #

需要实现与以下SQL逻辑完全一致的SQLAlchemy关联子查询：

SELECT id, ime, prezime, ratni_staz 
FROM stanovnik s, adresa a1 
WHERE s.sifra_adr = a1.sifra_adrese
  AND ratni_staz > (SELECT AVG(ratni_staz) 
                    FROM stanovnik s, adresa a2
                    WHERE s.sifra_adr = a2.sifra_adrese 
                      AND a1.sifra_adrese = a2.sifra_adrese)

注：修正了原SQL中子查询FROM子句的语法错误（原写法a2.sifra_adrese应为adresa a2）

实现方案 #

方式一：使用ORM实体别名 #

from sqlalchemy import select, func, and_

# 为表创建别名，对应SQL中的s、a1、s_sub、a2
s = Person.alias('s')
a1 = Address.alias('a1')
s_sub = Person.alias('s_sub')
a2 = Address.alias('a2')

# 构建关联子查询：计算当前地址的平均ratni_staz
subquery = select(func.avg(s_sub.ratni_staz)).where(
    and_(
        s_sub.sifra_adr == a2.sifra_adrese,
        a2.sifra_adrese == a1.sifra_adrese
    )
).scalar_subquery()

# 构建主查询：筛选出ratni_staz大于对应地址平均值的人员
stmt = select(
    s.id,
    s.ime,
    s.prezime,
    s.ratni_staz
).join(a1, s.sifra_adr == a1.sifra_adrese).where(
    s.ratni_staz > subquery
)

# 执行查询（假设已创建session）
results = session.execute(stmt).fetchall()

方式二：使用表对象别名（直接操作底层表结构） #

from sqlalchemy import select, func, and_

# 获取表对象并创建别名
s = Person.__table__.alias('s')
a1 = Address.__table__.alias('a1')
s_sub = Person.__table__.alias('s_sub')
a2 = Address.__table__.alias('a2')

# 关联子查询
subquery = select(func.avg(s_sub.c.ratni_staz)).where(
    and_(
        s_sub.c.sifra_adr == a2.c.sifra_adrese,
        a2.c.sifra_adrese == a1.c.sifra_adrese
    )
).scalar_subquery()

# 主查询
stmt = select(
    s.c.id,
    s.c.ime,
    s.c.prezime,
    s.c.ratni_staz
).select_from(
    s.join(a1, s.c.sifra_adr == a1.c.sifra_adrese)
).where(
    s.c.ratni_staz > subquery
)

# 执行查询
results = session.execute(stmt).fetchall()

代码说明 #

• scalar_subquery()将子查询转换为标量表达式，适合用于比较条件中
• 别名的使用完全对应目标SQL中的表别名，保证逻辑一致性
• 两种方式的核心逻辑一致，区别在于是否直接使用ORM实体或底层表对象，可根据项目需求选择

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