sqlalchemy版本:1.4.0
这是我想运行的postgres查询。
SELECT
"table_a".id,
"table_a".body,
"table_a".created_at,
"table_a".archived_at,
"user".full_name,
(
SELECT json_agg(items_alias_b)
FROM (
SELECT items_alias_a.id, items_alias_a.type, items_alias_a.filename, items_alias_a.s3_original_key
FROM items items_alias_a
WHERE items_alias_a.status = 'completed'::itemsstatusenum
AND items_alias_a.id = ANY("table_a".item_ids)
) items_alias_b
) AS "items"
FROM "table_a"
JOIN "user" ON "user".id = "table_a".created_by
WHERE "table_a".other_id = 'XXXXXXXX'
ORDER BY "table_a".id DESC
我正在尝试在SQLAlchemy中这样做,但不知道如何做到这一点。如果有人能帮上忙,那就太好了。
class TableA(Base):
__tablename__ = "table_a"
id = Column(BigInteger, primary_key=True)
other_id = Column(
String,
ForeignKey("other.id", use_alter=True, ondelete="CASCADE"),
index=True,
nullable=False,
)
body = Column(Text, index=True)
item_ids = Column(ARRAY(String))
created_at = Column(DateTime(timezone=False), server_default=func.now())
created_by = Column(
BigInteger,
ForeignKey(User.id, ondelete="CASCADE"),
nullable=False,
)
archived_at = Column(DateTime(timezone=False), nullable=True)
class Item(Base):
__tablename__ = "item"
id = Column(String, primary_key=True)
type = Column(String, nullable=False)
name = Column(String, nullable=False)
status = Column(String, nullable=False)
items_subquery = db_session.query(
Model.Item.id,
Model.Item.type,
Model.Item.name,
).filter(
Model.Item.status == Model.ItemStatusEnum.completed,
Model.Item.id == func.any(Model.TableA.items_ids),
).subquery()
query = (
db_session.query(
Model.TableA.id,
Model.TableA.body,
Model.TableA.created_at,
Model.TableA.archived_at,
Model.User.full_name.label("full_name"),
func.json_agg(items_subquery).label("items"),
)
.join(
Model.User,
Model.User.id == Model.TableA.created_by,
)
.filter(
Model.TableA.other_id == other_id,
)
.order_by(Model.TableA.id.desc())
)
messages = query.all()
当sqlalchemy运行时,我得到以下日志。这个查询与我预期的有点不同。我该怎么办?
(psycopg2.errors.SyntaxError) subquery must return only one column
SELECT
table_a.id AS table_a_id,
table_a.body AS table_a_body,
table_a.created_at AS table_a_created_at,
table_a.archived_at AS table_a_archived_at,
"user".full_name AS full_name,
(
SELECT
json_agg(
(
SELECT
item.id,
item.type,
item.name
FROM
item,
table_a
WHERE
item.status = %(status_1) s
AND item.id = any(table_a.item_ids)
)
) AS json_agg_1
) AS item
FROM
table_a
JOIN "user" ON "user".id = table_a.created_by
WHERE
table_a.other_id = %(other_id_1)
ORDER BY
table_a.id DESC
我试图从postgres查询中获得正确的sqlalchemy代码,但它不起作用,因为我在上面分享了。
1条答案
按热度按时间xv8emn3q1#
我看到你的帖子,同时试图解决与你相同的问题。经过长时间的斗争,我能够了解到底是怎么回事,并找到解决办法。似乎无法将子查询变量直接传递到select子句中。所以
func.json_agg(subquery)
会尝试编译成这样的东西:json_agg((SELECT ...))
在postgres中不允许。你需要做的是显式地告诉sqlalchemy,你想使用这些值而不是查询本身。对我来说有效的是调用子查询的
table_valued
方法,所以目标代码如下所示:func.json_agg(subquery.table_valued())
希望它也能为你工作。
PS这是我第一次偶然发现一个不到一个月的问题。