Converting a CTE query to SQLAlchemy ORM

Question

(This is a rewritten version of a deleted question from earlier today.)

I'm using the SQLAlchemy ORM as part of a Flask app, with MySQL as a backend, and I'm trying to write a query to return a list of entries surrounding a particular entry. While I have a working query in SQL, I'm unsure how to code it in SQLA. The docs for CTE in the ORM show a very complicated example, and there aren't many other examples I can find.

For now assume a very simple table that only contains words:

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

If I want the 10 words before and after a given word (with an id of 73), an SQL query that does what I need is:

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;

I can't figure out the next step, however. I want to get a list of Word objects. I'd imagine that the first part of it could be something like

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

but even if this is right, I don't know how to get this into the next part; I got bogged down in the aliased bits in the examples. Could someone give me a push in the right direction?

user_2144390 · Answer

This may not be the most elegant solution but it seems to be working for me:

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""


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:
[,
 ,
 ,
 ,
 ,
 ,
 ,
 ,
 ]
"""

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:
[,
 ,
 ,
 ,
 ,
 ]
 ]
"""

Converting a CTE query to SQLAlchemy ORM

1 Answers