sqlalchemy_bug.py

#!/usr/bin/env python2
# encoding: utf-8
from __future__ import print_function
import sqlalchemy as sqla
from sqlalchemy import orm, types
from sqlalchemy.sql.expression import bindparam
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy

metadata = sqla.MetaData()
Table = declarative_base(metadata=metadata)

class Foo(Table):
    __tablename__ = 'foo'
    id = sqla.Column(types.Integer, primary_key=True, nullable=False)

    def __init__(self, id):
        self.id = id

class FooName(Table):
    __tablename__ = 'foo_names'
    foo_id = sqla.Column(types.Integer, sqla.ForeignKey(Foo.id), nullable=False, primary_key=True)
    local_language_id = sqla.Column(types.Integer, nullable=False, primary_key=True)
    name = sqla.Column(types.Unicode, nullable=False)

    def __init__(self, foo_id, local_language_id, name):
        self.foo_id = foo_id
        self.local_language_id = local_language_id
        self.name = name

Foo.names_local = orm.relationship(FooName,
        primaryjoin=sqla.and_(
            FooName.foo_id == Foo.id,
            FooName.local_language_id == bindparam('_default_language_id',
                value='dummy', type_=types.Integer, required=True),
        ),
        foreign_keys=[FooName.foo_id, FooName.local_language_id],
        uselist=False,
)
Foo.name = association_proxy('names_local', 'name')

class MultilangQuery(orm.Query):
    def __iter__(self):
        if '_default_language_id' not in self._params or self._params['_default_language_id'] == 'dummy':
            self._params = self._params.copy()
            self._params['_default_language_id'] = self.session.default_language_id

        return super(MultilangQuery, self).__iter__()


class MultilangSession(orm.Session):
    """A tiny Session subclass that adds support for a default language.

    Needs to be used with `MultilangScopedSession`, below.
    """
    default_language_id = None

    def __init__(self, *args, **kwargs):
        if 'default_language_id' in kwargs:
            self.default_language_id = kwargs.pop('default_language_id')
        kwargs.setdefault('query_cls', MultilangQuery)
        super(MultilangSession, self).__init__(*args, **kwargs)

def test():
    ENGLISH_ID = 9
    engine = sqla.create_engine('sqlite:///:memory:', echo=True)
    sessionmaker = orm.sessionmaker(class_=MultilangSession, default_language_id=ENGLISH_ID)
    session = sessionmaker(bind=engine)

    # create tables
    metadata.create_all(engine)

    # populate the db
    session.add(Foo(1))
    session.add(FooName(1, 9, u'test'))
    session.add(FooName(1, 1, u'テスト'))
    session.commit()

    print("\n"*4)

    # get some entries
    foo = session.query(Foo).one()
    print(repr(foo.name))
    print("\n"*4)
    del foo

    foo = session.query(Foo).options(orm.joinedload(Foo.names_local)).one()
    print(repr(foo.name))
    print("\n"*4)
    del foo

    print("SQLAlchemy", sqla.__version__)

test()