primaryjoin with bindparam test case

testcase.py

from sqlalchemy import Column, Integer, ForeignKey, bindparam, and_, create_engine
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


def get_age():
    """Method that would get some data from a Flask request context, although
    the behaviour is the same regardless."""
    return 15


class Child(Base):
    __tablename__ = 'child'
    id = Column('id', Integer, primary_key=True)
    age = Column(Integer)
    parent_id = Column(Integer, ForeignKey('parent.id'))


class Parent(Base):
    __tablename__ = 'parent'
    id = Column('id', Integer, primary_key=True)
    @classmethod
    def __declare_last__(cls):
        cls.children_by_age = relationship(
            Child,
            lazy='joined',
            primaryjoin=and_(
                Parent.id == Child.parent_id,
                Child.age == bindparam('age', callable_=get_age)
            )
        )

### DB setup ###

engine = create_engine('sqlite://')
session = sessionmaker(bind=engine)()
Base.metadata.create_all(engine)

parent = Parent()
session.add(parent)
session.commit()

c1, c2, c3 = [
    Child(age=10, parent_id=parent.id),
    Child(age=15, parent_id=parent.id),
    Child(age=20, parent_id=parent.id)
]

session.add_all([c1, c2, c3])
session.commit()

### Test case that shows odd behaviour ###

assert len(parent.children_by_age) == 0 # not there, unsurprisingly

session.expunge_all()

parent = session.query(Parent).one()
assert len(parent.children_by_age) == 1 # the Child object is now loaded

session.add(parent)
session.commit()
assert len(parent.children_by_age) == 0 # the Child object is gone!
session.refresh(parent)
assert len(parent.children_by_age) == 1 # it's back again

testcase_with_flask.py

from flask import Flask, has_request_context
from sqlalchemy import Column, Integer, ForeignKey, bindparam, and_, create_engine
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


def get_age():
    """Method that would get some data from a Flask request context, although
    the behaviour is the same regardless."""
    if has_request_context():
        return 15


class Child(Base):
    __tablename__ = 'child'
    id = Column('id', Integer, primary_key=True)
    age = Column(Integer)
    parent_id = Column(Integer, ForeignKey('parent.id'))


class Parent(Base):
    __tablename__ = 'parent'
    id = Column('id', Integer, primary_key=True)
    @classmethod
    def __declare_last__(cls):
        cls.children_by_age = relationship(
            Child,
            lazy='joined',
            primaryjoin=and_(
                Parent.id == Child.parent_id,
                Child.age == bindparam('age', callable_=get_age)
            )
        )

### DB setup ###

app = Flask(__name__)

engine = create_engine('sqlite://')
session = sessionmaker(bind=engine)()
Base.metadata.create_all(engine)

parent = Parent()
session.add(parent)
session.commit()

c1, c2, c3 = [
    Child(age=10, parent_id=parent.id),
    Child(age=15, parent_id=parent.id),
    Child(age=20, parent_id=parent.id)
]

session.add_all([c1, c2, c3])
session.commit()

### Test case that shows odd behaviour ###

assert len(parent.children_by_age) == 0 # not there, unsurprisingly

session.expunge_all()

parent = session.query(Parent).one()
assert len(parent.children_by_age) == 0 # the Child object is not there, no request context

with app.test_request_context():
    parent = session.query(Parent).one()
    assert len(parent.children_by_age) == 0 # still not there, needs refresh
    session.refresh(parent) # expire + new query will have the same result
    assert len(parent.children_by_age) == 1 # now it works
    session.add(parent)
    session.commit()
    assert len(parent.children_by_age) == 0 # gone after commit
    session.refresh(parent)
    assert len(parent.children_by_age) == 1 # back again