如何将原生PostgreSQL查询转换为SQLAlchemy？子查询关联问题求助

原生PostgreSQL查询 #

SELECT a.main.id, a.main.name, a.access.user_id,
(select count(*) from a.access where a.access.main_id = a.main.id and a.access.is_admin IS FALSE) as count
FROM a.main
INNER JOIN a.access ON a.access.main_id = a.main.id
where a.access.user_id = 'myusername' 
group by a.main.id, a.main.name, a.access.user_id

已编写的SQLAlchemy代码 #

subquery = session.query(
    func.count().label("count")
).filter(
    Access.main_id == Main.id,
    Access.is_admin.is_(False),
).subquery()

stmt = session.query(
    Main.id,
    Main.name,
    Access.user_id,
    subquery.c.count
).join(
    Access,
    Access.main_id == Main.id
).filter(
    Access.user_id == 'myusername'
).group_by(
    Main.id,
    Main.name,
    Access.user_id,
    subquery.c.count
).all()

遇到的问题 #

生成的子查询不符合预期，自动将a.main加入了子查询的FROM子句：

SELECT count(*)
FROM a.access, a.main
WHERE a.access.main_id = a.main.id

解决方案 #

问题出在子查询是相关子查询，需要明确告知SQLAlchemy子查询中引用的Main表来自外部查询，而非子查询自身的FROM子句。可以通过correlate()方法实现：

修正后的代码 #

# 定义相关子查询，指定关联外部的Main表
subquery = session.query(
    func.count().label("count")
).filter(
    Access.main_id == Main.id,
    Access.is_admin.is_(False),
).correlate(Main)  # 关键：指定该表来自外部查询，不加入子查询的FROM
.subquery()

stmt = session.query(
    Main.id,
    Main.name,
    Access.user_id,
    subquery.c.count
).join(
    Access,
    Access.main_id == Main.id
).filter(
    Access.user_id == 'myusername'
).group_by(
    Main.id,
    Main.name,
    Access.user_id
).all()

说明 #

1. correlate(Main)：告诉SQLAlchemy，子查询中使用的Main表是外部查询的表，不会将其添加到子查询的FROM列表中，从而生成正确的相关子查询。
2. 移除group_by中的subquery.c.count：因为count是子查询的结果，不属于分组字段，原原生SQL也未将其纳入GROUP BY。
3. 修正语法错误：字段间添加逗号，条件判断使用==而非=（Python语法要求）。

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