broken_autojoin.py

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

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

Session = sessionmaker(engine)


def get_primaryjoin():

    def _get_lang_id():
        return languages['de']
    lang_id = bindparam('language_id', callable_=_get_lang_id)
    return and_(
        Article.id == ArticleI18N.article_id,
        ArticleI18N.language_id == lang_id,
    )


class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True)
    german = relationship(
        'ArticleI18N', uselist=False,
        primaryjoin=get_primaryjoin,
        viewonly=True, lazy='joined', innerjoin=True,
    )


class Language(Base):
    __tablename__ = 'language'
    id = Column(Integer, primary_key=True)
    code = Column(Unicode)


class ArticleI18N(Base):
    __tablename__ = 'article_i18n'
    article_id = Column(ForeignKey('article.id'), primary_key=True)
    language_id = Column(ForeignKey('language.id'), primary_key=True)
    title = Column(Unicode)

    article = relationship(Article, backref='translations')
    language = relationship(Language)


Base.metadata.create_all(engine)


session = Session()
en = Language(code='en')
de = Language(code='de')
session.add_all([en, de])
session.flush()
languages = {
    'de': de.id,
    'en': en.id,
}


article = Article()
article.translations.append(ArticleI18N(title='Title EN', language=en))
article.translations.append(ArticleI18N(title='Title DE', language=de))
session.add(article)
session.commit()
session.close()

session = Session()
article = session.query(Article).one()
print('Title: %s' %article.german.title)
session.expire(article)
print('Title: %s' %article.german.title)