将 CTE 查询转换为 SQLAlchemy ORM答案

【问题标题】：Converting a CTE query to SQLAlchemy ORM将 CTE 查询转换为 SQLAlchemy ORM
【发布时间】：2020-10-14 18:16:52
【问题描述】：

（这是今天早些时候删除的问题的重写版本。）

我使用 SQLAlchemy ORM 作为 Flask 应用程序的一部分，使用 MySQL 作为后端，我正在尝试编写一个查询以返回围绕特定条目的条目列表。虽然我在 SQL 中有一个有效的查询，但我不确定如何在 SQLA 中对其进行编码。 docs for CTE in the ORM 给出了一个非常复杂的例子，其他的例子我找不到了。

现在假设一个非常简单的表，只包含单词：

class Word(db.Model):
    __tablename__ = 'word'
    id = db.Column(db.Integer, primary_key=True)
    word = db.Column(db.String(100))

如果我想要给定单词之前和之后的 10 个单词（id 为 73），那么满足我需要的 SQL 查询是：

WITH cte AS (SELECT id, word, ROW_NUMBER() OVER (ORDER BY word) AS rownumber FROM word)
SELECT * FROM cte
WHERE rownumber > (SELECT rownumber FROM cte WHERE cte.id = 73) - 10
AND rownumber < (SELECT rownumber FROM cte WHERE cte.id = 73) + 10
ORDER BY rownumber;

但是，我无法弄清楚下一步。我想获取 Word 对象的列表。我想它的第一部分可能是这样的

id = 73
rowlist = db.session.query(Word.id, db.func.row_number()).filter(Word.id == id).order_by(Word.word).cte()

但即使这是正确的，我也不知道如何将它带入下一部分；我陷入了示例中的aliased 位。有人可以推动我朝着正确的方向前进吗？

【问题讨论】：

标签： python sql sqlalchemy

【解决方案1】：

这可能不是最优雅的解决方案，但它似乎对我有用：

engine = db.create_engine(sqlalchemy_uri)

Base = declarative_base()


class Word(Base):
    __tablename__ = "so64359277"
    id = db.Column(db.Integer, primary_key=True)
    word = db.Column(db.String(100))

    def __repr__(self):
        return f"<Word(id={self.id}, word='{self.word}')>"


Base.metadata.drop_all(engine, checkfirst=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# test data
word_objects = []
for x in [
    "Hotel",
    "Charlie",
    "Alfa",
    "India",
    "Foxtrot",
    "Echo",
    "Bravo",
    "Golf",
    "Delta",
]:
    word_objects.append(Word(word=x))
session.add_all(word_objects)
session.commit()
# show test data with id values
pprint(session.query(Word).all())
"""console output:
[<Word(id=1, word='Hotel')>,
 <Word(id=2, word='Charlie')>,
 <Word(id=3, word='Alfa')>,
 <Word(id=4, word='India')>,
 <Word(id=5, word='Foxtrot')>,
 <Word(id=6, word='Echo')>,
 <Word(id=7, word='Bravo')>,
 <Word(id=8, word='Golf')>,
 <Word(id=9, word='Delta')>]
"""

target_word = "Echo"
num_context_rows = 3

rowlist = session.query(
    Word.id,
    Word.word,
    db.func.row_number().over(order_by=Word.word).label("rownum"),
).cte("rowlist")
target_rownum = session.query(rowlist.c.rownum).filter(
    rowlist.c.word == target_word
)
select_subset = session.query(rowlist.c.rownum, rowlist.c.id).filter(
    db.and_(
        (rowlist.c.rownum >= target_rownum.scalar() - num_context_rows),
        (rowlist.c.rownum <= target_rownum.scalar() + num_context_rows),
    )
)
rownum_id_map = {x[0]: x[1] for x in select_subset.all()}
min_rownum = min(rownum_id_map)
max_rownum = max(rownum_id_map)
result = []
for rownum in range(min_rownum, max_rownum + 1):
    result.append(session.query(Word).get(rownum_id_map[rownum]))
pprint(result)
"""console output:
[<Word(id=7, word='Bravo')>,
 <Word(id=2, word='Charlie')>,
 <Word(id=9, word='Delta')>,
 <Word(id=6, word='Echo')>,
 <Word(id=5, word='Foxtrot')>,
 <Word(id=8, word='Golf')>]
 <Word(id=1, word='Hotel')>]
"""

【讨论】：