如何编写带有联接和聚合的 SQLAlchemy 查询?

Posted

技术标签:

【中文标题】如何编写带有联接和聚合的 SQLAlchemy 查询?【英文标题】:How can I write an SQLAlchemy Query with a Join and an Aggregate? 【发布时间】:2012-11-13 18:33:25 【问题描述】:

我有一个包含 3 列的表:类型、内容和时间(一个整数)。对于每个“类型”,我想选择具有最大(最近)“时间”整数和相应数据的条目。如何使用 SQLAlchemy 和 Python 做到这一点?我可以通过执行以下操作使用 SQL 来做到这一点:

select
  c.type,
  c.time,
  b.data
from
  parts as b

inner join

  (select
    a.type,
    max(a.time) as time
  from parts as a
  group by a.type) as c

on

b.type = c.type and
b.time = c.time

但是我怎样才能在 SQLAlchemy 中做到这一点呢?

表映射:

class Structure(Base):
    __tablename__ = 'structure'
    id = Column(Integer, primary_key=True)
    type = Column(Text)
    content = Column(Text)
    time = Column(Integer)

    def __init__(self, type, content):
        self.type = type
        self.content = content
        self.time = time.time()

    def serialise(self):
        return "type" : self.type,
            "content" : self.content;

尝试的查询:

    max = func.max(Structure.time).alias("time")
    c = DBSession.query(max)\
        .add_columns(Structure.type, Structure.time)\
        .group_by(Structure.type)\
        .subquery()
    c.alias("c")

    b = DBSession.query(Structure.content)\
        .add_columns(c.c.type, c.c.time)\
        .join(c, Structure.type == c.c.type)

给我:

sqlalchemy.exc.OperationalError: (OperationalError) 靠近 "(": 语法 错误 u'SELECT structure.content AS structure_content, anon_1.type AS anon_1_type, anon_1.time AS anon_1_time \nFROM 结构 JOIN (SELECT time.max_1 AS max_1, structure.type AS 类型, structure.time AS 时间 \nFROM max(structure.time) AS 时间,结构 GROUP BY structure.type) AS anon_1 ON structure.type = anon_1.type' ()

我基本上是在黑暗中刺伤,所以任何帮助将不胜感激。

【问题讨论】:

这可以使用 SQLAlchemy 来完成。你已经尝试过什么了吗?您是否定义了任何表映射? @MartijnPieters 嘿 Martijn,我昨晚在离开办公室的路上发布了这篇文章。一旦我有机会再看一遍,我会添加一些代码;谢谢。 @MartijnPieters 到目前为止我已经添加了我的工作,但不知道我是否走在正确的轨道上。 【参考方案1】:

使用子查询试试下面的代码:

subq = (session.query(
            Structure.type, 
            func.max(Structure.time).label("max_time")
        ).
        group_by(Structure.type)
        ).subquery()

qry = (session.query(Structure).
       join(subq, and_(Structure.type == subq.c.type, Structure.time == subq.c.max_time))
       )

print qry

生成 SQL:

SELECT  structure.id AS structure_id, structure.type AS structure_type, structure.content AS structure_content, structure.time AS structure_time
FROM    structure 
JOIN    (SELECT structure.type AS type, max(structure.time) AS max_time
         FROM structure GROUP BY structure.type) AS anon_1 
    ON  structure.type = anon_1.type 
    AND structure.time = anon_1.max_time

【讨论】:

赢家!感谢您的帮助,您已经解决了我的主要问题以及一些较小的问题(例如“我如何加入多个专栏?”),非常感谢您的帮助;谢谢。

以上是关于如何编写带有联接和聚合的 SQLAlchemy 查询?的主要内容,如果未能解决你的问题,请参考以下文章

sqlalchemy中的外连接

进行联接时如何更改 SqlSoup / SqlAlchemy 中的 FROM 子句?

如何将 Sqlalchemy ORM 查询结果转换为包含关系的单个联接表?

如何在单个查询中使用联接和聚合函数更新表中的多行

按 sqlalchemy 中的连接别名排序

二级联接Sqlalchemy