python - 将 CTE 查询转换为 SQLAlchemy ORM

Question

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

我正在使用 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 位。有人能给我一个正确的方向吗？

Answer 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')>]
"""