recommended way to eagerly load object relationship retrieved via complex query ?

[gerazenobi]

Hi there,

After going through the documentation for a while, I was wondering what would be the best/recommended way to construct an eagerly loaded relationship that would involve a rather difficult query/subquery

Let's say I have the following situation: a task which have multiple task updates and I want to create a joined relationship on the task object to always have the first task update loaded along with the task.

import datetime
from datetime import timedelta

import pytz
from sqlalchemy import create_engine, Column, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.types import DateTime, Integer, String

engine = create_engine("sqlite://", echo=True)
Base = declarative_base()


def timestamp_utc_with_timezone():
    dt = datetime.datetime.utcnow()
    dt = dt.replace(tzinfo=pytz.utc)
    return dt


class Task(Base):
    __tablename__ = "task"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)

    updates = relationship("TaskUpdate", passive_deletes="all")

    # first_update = relationship("TaskUpdate", lazy="joined") # HERE

    def __repr__(self):
        return f"Task(id={self.id})"


class TaskUpdate(Base):
    __tablename__ = "task_update"

    id = Column(Integer, primary_key=True)
    task_id = Column(Integer, ForeignKey("task.id", ondelete="CASCADE"))
    timestamp = Column(DateTime(timezone=True), default=timestamp_utc_with_timezone)
    info = Column(String)

    def __repr__(self):
        return f"TaskUpdate(task_id={self.task_id}, timestamp={self.timestamp.isoformat()})"


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
task = Task(name="Task 1")
session.add(task)
for i in range(5):
    st = TaskUpdate(info=f"info:{i}", timestamp=timestamp_utc_with_timezone() - timedelta(hours=1))
    task.updates.append(st)
session.commit()
task = session.query(Task).first()
# print(task.first_update)

in SQL first_update could be retrieved for instance as:

-- example:

SELECT task.*, task_update.*
FROM (
         SELECT task_1.*,
                (SELECT task_update.id
                 FROM task_update
                 where task_update.task_id = task_1.id
                 ORDER BY timestamp
                 LIMIT 1) first_task_update_id
         FROM task As task_1
         where task_1.id = 1
     ) task
         JOIN task_update as task_update ON task_update.id = task.first_task_update_id;

Is this even a use case for a relationship or too complex ?

Thanks in advance for any advice.

[zzzeek]

Hi -

you can map to that limit query using the "relationship to aliased class" feature, then make a relationship to that. If your database supports window functions (all the major DBs do now), you can use the recipe at https://docs.sqlalchemy.org/en/14/orm/join_conditions.html#row-limited-relationships-with-window-functions directly which will allow all kinds of eager loading such as "selectinload" to work as expected.

[gerazenobi]

Hi @zzzeek , thanks for the quick reply :)

Following your suggestion then I did:

from sqlalchemy import create_engine, Column, ForeignKey, func, and_, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, aliased
from sqlalchemy.types import DateTime, Integer, String

# ...

partitioned_task_updates = select(
    TaskUpdate,
    func.row_number().over(order_by=TaskUpdate.timestamp, partition_by=TaskUpdate.task_id).label("update_number"),
).alias()

ordered_updates = aliased(TaskUpdate, partitioned_task_updates)

Task.first_update = relationship(
    ordered_updates,
    primaryjoin=and_(ordered_updates.task_id == Task.id, partitioned_task_updates.c.update_number == 1),
)

#... 
print(task.first_update)

I got the following error:

Traceback (most recent call last):
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/pydevd.py", line 1477, in _exec
    pydev_imports.execfile(file, globals, locals)  # execute the script
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_imps/_pydev_execfile.py", line 18, in execfile
    exec(compile(contents+"\n", file, 'exec'), glob, loc)
  File "/Users/gerardozenobi/Library/Application Support/JetBrains/PyCharm2020.3/scratches/alchemy_question.py", line 52, in <module>
    func.row_number().over(order_by=TaskUpdate.timestamp, partition_by=TaskUpdate.task_id).label("update_number"),
  File "<string>", line 2, in select
  File "<string>", line 2, in __init__
  File "/Users/gerardozenobi/workspaces/bpr-api/env/lib/python3.6/site-packages/sqlalchemy/util/deprecations.py", line 139, in warned
    return fn(*args, **kwargs)
  File "/Users/gerardozenobi/workspaces/bpr-api/env/lib/python3.6/site-packages/sqlalchemy/sql/selectable.py", line 3114, in __init__
    for c in columns:
TypeError: 'DeclarativeMeta' object is not iterable

I can't seem to spot what I did wrong. Any ideas?

Thanks again.

[zzzeek]

it looks like you're on an older SQLAlchemy - the example above is using 1.4 style select(). If you are on SQLAlchemy 1.3, add brackets to the select(): select([TaskUpdate, func.row_number(), ...]). the pattern requires at least sqlalchemy 1.3 also.

[gerazenobi]

Right, it was that thanks !

I got another error nonetheless but I suspect it is due to SQLITE ?

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "(": syntax error

Since the SQL constructed seems to be fine:

SELECT anon_1.id        AS anon_1_id,
       anon_1.task_id   AS anon_1_task_id,
       anon_1.timestamp AS anon_1_timestamp,
       anon_1.info      AS anon_1_info
FROM (SELECT task_update.id                                                                      AS id,
             task_update.task_id                                                                 AS task_id,
             task_update.timestamp                                                               AS timestamp,
             task_update.info                                                                    AS info,
             row_number() OVER (PARTITION BY task_update.task_id ORDER BY task_update.timestamp) AS update_number
      FROM task_update) AS anon_1
WHERE anon_1.task_id = ?
  AND anon_1.update_number = ?;

and when running it directly against a postgres database it run without issues 👍

Any ideas for the SQLite error ? (just curious to know what you might know about it)

Anyway thanks for your help @zzzeek, I have a working example for postgres 🚀

[zzzeek]

older SQLite's didn't have window functions can you try this:

you need at least 3.25.0

$ python
Python 3.9.2 (default, Feb 20 2021, 00:00:00) 
[GCC 10.2.1 20201125 (Red Hat 10.2.1-9)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import sqlite3
>>> sqlite3.sqlite_version
'3.34.1'
>>> 

[gerazenobi]

That must be it then, I got 3.24.0 as we are currently using python 3.6

Thanks for the info again!