Late evaluation of "secondaryjoin"

[nmoreaud]

Hello

When a secondaryjoin uses an aliased query, I wonder how I can use late evaluation of the relationship.
Here is an example copy/pasted from stackoverflow. how can I late evaluate friendship_union, primaryjoin and secondaryjoin?
My real use case uses multiple tables.
Thanks in advance

from sqlalchemy import Integer, Table, Column, ForeignKey, \
    create_engine, String, select
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.declarative import declarative_base

Base= declarative_base()

friendship = Table(
    'friendships', Base.metadata,
    Column('friend_a_id', Integer, ForeignKey('users.id'), 
                                        primary_key=True),
    Column('friend_b_id', Integer, ForeignKey('users.id'), 
                                        primary_key=True)
)


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    # this relationship is used for persistence
    friends = relationship("User", secondary=friendship, 
                           primaryjoin=id==friendship.c.friend_a_id,
                           secondaryjoin=id==friendship.c.friend_b_id,
    )

    def __repr__(self):
        return "User(%r)" % self.name

# this relationship is viewonly and selects across the union of all
# friends
friendship_union = select([
                        friendship.c.friend_a_id, 
                        friendship.c.friend_b_id
                        ]).union(
                            select([
                                friendship.c.friend_b_id, 
                                friendship.c.friend_a_id]
                            )
                    ).alias()
User.all_friends = relationship('User',
                       secondary=friendship_union,
                       primaryjoin=User.id==friendship_union.c.friend_a_id,
                       secondaryjoin=User.id==friendship_union.c.friend_b_id,
                       viewonly=True) 

e = create_engine("sqlite://",echo=True)
Base.metadata.create_all(e)
s = Session(e)

u1, u2, u3, u4, u5 = User(name='u1'), User(name='u2'), \
                    User(name='u3'), User(name='u4'), User(name='u5')

u1.friends = [u2, u3]
u4.friends = [u2, u5]
u3.friends.append(u5)
s.add_all([u1, u2, u3, u4, u5])
s.commit()

print u2.all_friends
print u5.all_friends

[CaselIT]

Hi,

I would suggest you start by looking at the documentation https://docs.sqlalchemy.org/en/20/orm/relationship_api.html#sqlalchemy.orm.relationship.params.primaryjoin

[nmoreaud]

Okay, I have found a way, probably not the most elegant one but it seems to work.
I store the aliased join in a cached static method, and reuse it in multiple lambdas:

from functools import cache
# ...

class User(Base):
    # ...

    @cache
    @staticmethod
    def _friendship_union():
        return (
            select(friendship.c.friend_a_id, friendship.c.friend_b_id)
        ).union(
            select(friendship.c.friend_b_id, friendship.c.friend_a_id)
        ).alias()

    all_friends = relationship(
        'User',
        secondary=lambda: User._friendship_union(),
        primaryjoin=lambda: User.id == User._friendship_union().c.friend_a_id,
        secondaryjoin=lambda: User.id == User._friendship_union().c.friend_b_id,
        viewonly=True)

[CaselIT]

you can also look at declared_attr to see if it helps

[nmoreaud]

No, declared_attr doesn't seem to work here.
But I can make a custom decorator:

def lateinit_secondaryjoin(func):
    """Decorator to declare a late initialized secondaryjoin condition when every model is not defined yet."""
    return cache(staticmethod(func))

Thanks

[CaselIT]

Depending on where the friendship table is defied i meant something like

@declared_attr 
def all_friends(cls) -> Mapped[User]:
    subquery = select(friendship.c.friend_a_id, friendship.c.friend_b_id).union(
      select(friendship.c.friend_b_id, friendship.c.friend_a_id)).alias()
    return relationship(
        'User',
        secondary=subquery,
        primaryjoin=cls.id == subquery.c.friend_a_id,
        secondaryjoin=cls.id == subquery.c.friend_b_id,
        viewonly=True)

[nmoreaud]

Yes, this would be very nice, but there seem to be a side effect, because it raises sqlalchemy.exc.NoReferencedTableError: Foreign key associated with column 'anon_1.friend_a_id' could not find table 'users' with which to generate a foreign key to target column 'id' as soon as all_friends is defined.

I have fixed the example code:

from sqlalchemy import Integer, Column, ForeignKey, String, select, create_engine
from sqlalchemy.orm import relationship, declared_attr, Mapped, declarative_base, Session

Base = declarative_base()


class Friendship(Base):
    __tablename__ = 'friendships'

    friend_a_id = Column(Integer, ForeignKey('users.id'), primary_key=True)
    friend_b_id = Column(Integer, ForeignKey('users.id'), primary_key=True)


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    # this relationship is used for persistence
    friends = relationship(
        "User",
        secondary=Friendship.__table__,
        primaryjoin=id == Friendship.friend_a_id,
        secondaryjoin=id == Friendship.friend_b_id
    )

    def __repr__(self):
        return "User(%r)" % self.name

    @declared_attr
    def all_friends(cls) -> Mapped['User']:
        subquery = select(Friendship.friend_a_id, Friendship.friend_b_id).union(
            select(Friendship.friend_b_id, Friendship.friend_a_id)).alias()
        return relationship(
            'User',
            secondary=subquery,
            primaryjoin=cls.id == subquery.c.friend_a_id,
            secondaryjoin=cls.id == subquery.c.friend_b_id,
            viewonly=True)


e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)

u1, u2, u3, u4, u5 = User(name='u1'), User(name='u2'), \
    User(name='u3'), User(name='u4'), User(name='u5')

u1.friends = [u2, u3]
u4.friends = [u2, u5]
u3.friends.append(u5)
s.add_all([u1, u2, u3, u4, u5])
s.commit()

print(u2.all_friends)
print(u5.all_friends)

[CaselIT]

Yes I think the issue is that declared_attr is resolved before creating the user table, so in this case you still need use late evaluation and your solution is likely the right one

[zzzeek]

you're using Python 2 with what SQLAlchemy version? your program at the top runs without issue in SQLAlchemy 1.4 with Python 3, there's nothing I can see wrong with it

[zzzeek]

if the question is "how can I have this inline with Friends" you'd need to use an event hook like declare_last

[nmoreaud]

I think that I have the latest version of SQLAlchemy 2, I may be a little outdated.
With a cached function the code works. With declared_attr the syntax is better but it throws an error.