PoC to have sqlalchemy relationship backpopulate correctly even with custom primaryjoin condition

test_sqla_relations_backpopulate_primaryjoin.py

def test_relationships_backpopulate_primaryjoin():
    Base = declarative_base()

    class Post(Base):
        __tablename__ = 'post'

        id = Column(Integer, primary_key=True, autoincrement=True)

        text = Column(String(255))
        active = Column(Boolean())
        user_id = Column(Integer, ForeignKey('user.id'), nullable=False)
        user = relationship('User', back_populates='posts')

        def __unicode__(self):
            return 'Post {self.id} - {self.active}'.format(self=self)

        def __repr__(self):
            return unicode(self)


    @event.listens_for(Post.active, 'set')
    def hook_active(post, post_isactive, oldvalue, initiator):
        if post.user is None or post_isactive is None:
            return
        set_user_backref(post=post, user=post.user, is_active=post_isactive)

    @event.listens_for(Post.user, 'set')
    def hook_user(post, post_user, oldvalue, initiator):
        if post_user is None or post.active is None:
            return
        set_user_backref(post=post, user=post_user, is_active=post.active)

    def set_user_backref(post, user, is_active):
        if is_active:
            if post not in user.active_posts:
                user.active_posts.append(post)
            if post in user.inactive_posts:
                user.inactive_posts.remove(post)
        else:
            if post not in user.inactive_posts:
                user.inactive_posts.append(post)
            if post in user.active_posts:
                user.active_posts.remove(post)


    class User(Base):
        __tablename__ = 'user'

        id = Column(Integer, primary_key=True, autoincrement=True)

        name = Column(String(255))

        posts = relationship(Post, back_populates='user')
        active_posts = relationship(
            Post,
            primaryjoin=(Post.user_id == id) & Post.active,
            condition=lambda post: post.active,
            conditino_deps=Post.active,
            # condition=Post.active
        )
        inactive_posts = relationship(
            Post,
            # primaryjoin=(Post.user_id == id) & ~Post.active,
            # condition=lambda post: not post.active,
            condition=~Post.active,
        )

    engine = create_engine('sqlite:///:memory:', echo=True)
    configure_mappers()
    Base.metadata.create_all(engine)

    session = sessionmaker(bind=engine)()

    u = User(name='Alessio')
    p1 = Post(text='active', active=True, user=u)
    p2 = Post(text='inactive', active=False, user=u)
    make_assertions(session, u, p1, p2)
    session.rollback()

    u = User(name='Alessio')
    p1 = Post(text='active', active=False)
    p2 = Post(text='inactive', active=True)
    u.posts = [p1, p2]
    p1.active = True
    p2.active = False
    make_assertions(session, u, p1, p2)
    session.rollback()


def make_assertions(session, u, p1, p2):
    assert u.posts == [p1, p2]
    assert u.active_posts == [p1]
    assert u.inactive_posts == [p2]

    session.add(u)
    # session.flush()

    assert u.posts == [p1, p2]
    assert u.active_posts == [p1]
    assert u.inactive_posts == [p2]

    session.flush()

    assert u.posts == [p1, p2]
    assert u.active_posts == [p1]
    assert u.inactive_posts == [p2]