SQLAlchemy连接子查询时触发NotImplementedError问题排查

问题场景 #

尝试将ORM查询对象与原生SQL子查询进行连接，两个查询单独运行均正常，但执行连接操作时抛出NotImplementedError。

代码示例 #

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

from project import models

SQLALCHEMY_DATABASE_URL = "XXX"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionDB = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

session = SessionDB()
query = (
    session.query(
        models.Homes,
        models.RentalData,
        models.Descriptions,
    )
    .filter(
        models.Homes.id == models.RentalData.home_id,
    )
    .filter(
        models.Homes.id == models.Descriptions.home_id,
    )
)

# 原报错的子查询定义
query_2 = (
    session.query(models.Homes)
    .from_statement(
        text(
            """
    SELECT 
        id,
        (SELECT 
            ARRAY(
                    SELECT image_url 
                    FROM listings.images 
                    WHERE listings.images.home_id = listings.homes.id))
            AS image_url 
    FROM listings.homes"""
        )
    )
    .subquery()
)

query = query.join(query_2, models.Homes.id == query_2.c.id)

报错回溯信息 #

Traceback (most recent call last):
  File "/home/user/app/merge_example.py", line 52, in <module>
    query = query.join(query_2, models.Listings.id == query_2.c.id)
                                                      ^^^^^^^^^
  File "/home/user/app/env/lib/python3.11/site-packages/sqlalchemy/util/langhelpers.py", line 1113, in __get__
    obj.__dict__[self.__name__] = result = self.fget(obj)
                                           ^^^^^^^^^^^^^^
  File "/home/user/app/env/lib/python3.11/site-packages/sqlalchemy/sql/selectable.py", line 737, in columns
    self._populate_column_collection()
  File "/home/user/app/env/lib/python3.11/site-packages/sqlalchemy/sql/selectable.py", line 1643, in _populate_column_collection
    self.element._generate_fromclause_column_proxies(self)
  File "/home/user/app/env/lib/python3.11/site-packages/sqlalchemy/sql/selectable.py", line 3044, in _generate_fromclause_column_proxies
    raise NotImplementedError()
NotImplementedError

使用环境 #

• Postgres 11
• Python 3.11
• SQLAlchemy 1.4

原因分析 #

报错核心原因是：用session.query(models.Homes).from_statement(text(...))生成的ORM查询无法被SQLAlchemy正确转换为可连接的子查询。

这种方式是将原生SQL的结果映射到ORM模型，但当调用.subquery()时，SQLAlchemy无法从该ORM查询中提取出子查询所需的列元数据（比如列名、类型等），因此在访问query_2.c.id时触发了NotImplementedError。

解决方案 #

需要修改子查询的定义方式，明确告诉SQLAlchemy子查询的列信息，有两种常用方法：

方法1：用select结合显式列定义构造子查询 #

from sqlalchemy import select, column, text

# 替换原query_2的定义
query_2 = select(
    column("id"),
    column("image_url")
).from_statement(text("""
    SELECT 
        id,
        (SELECT ARRAY(SELECT image_url FROM listings.images WHERE listings.images.home_id = listings.homes.id)) AS image_url 
    FROM listings.homes
""")).subquery()

方法2：直接给text对象声明列信息 #

from sqlalchemy import text, column

# 替换原query_2的定义
query_2 = text("""
    SELECT 
        id,
        (SELECT ARRAY(SELECT image_url FROM listings.images WHERE listings.images.home_id = listings.homes.id)) AS image_url 
    FROM listings.homes
""").columns(
    id=column("id"),
    image_url=column("image_url")
).subquery()

修改后，SQLAlchemy能识别子查询的列，query_2.c.id可以正常访问，连接操作也能顺利执行。

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