from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm.scoping import ScopedSession
from sqlalchemy.orm.session import Session, object_session
+from sqlalchemy.engine.base import Connection
from sqlalchemy.schema import Column, ForeignKey, Table
-from sqlalchemy.sql.expression import and_, bindparam, select
+from sqlalchemy.sql.expression import and_, bindparam, select, Select
from sqlalchemy.types import Integer
def create_translation_table(_table_name, foreign_class, relation_name,
# Create the table object
table = Table(_table_name, foreign_class.__table__.metadata,
Column(foreign_key_name, Integer, ForeignKey(foreign_class.id),
- primary_key=True, nullable=False),
+ primary_key=True, nullable=False,
+ info=dict(description="ID of the %s these texts relate to" % foreign_class.__singlename__)),
Column('local_language_id', Integer, ForeignKey(language_class.id),
- primary_key=True, nullable=False),
+ primary_key=True, nullable=False,
+ info=dict(description="Language these texts are in")),
)
Translations.__table__ = table
'foreign_id': synonym(foreign_key_name),
'local_language': relationship(language_class,
primaryjoin=table.c.local_language_id == language_class.id,
- innerjoin=True),
+ innerjoin=True,
+ lazy='joined'),
})
# Add full-table relations to the original class
return Translations
class MultilangSession(Session):
- """A tiny Session subclass that adds support for a default language."""
- _default_language_id = 9 # English. XXX magic constant
+ """A tiny Session subclass that adds support for a default language.
- @property
- def default_language(self):
- # XXX need to get the right mapped class for this to work
- raise NotImplementedError
-
- @default_language.setter
- def default_language(self, new):
- self._default_language_id = new#.id
-
- @default_language.deleter
- def default_language(self):
- try:
- del self._default_language_id
- except AttributeError:
- pass
-
- def execute(self, clause, params=None, *args, **kwargs):
- if not params:
- params = {}
- params.setdefault('_default_language_id', self._default_language_id)
- return super(MultilangSession, self).execute(
- clause, params, *args, **kwargs)
+ 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')
+
+ super(MultilangSession, self).__init__(*args, **kwargs)
+
+ def connection(self, *args, **kwargs):
+ """Monkeypatch the connection. Not pretty at all.
+ """
+ conn = super(MultilangSession, self).connection(*args, **kwargs)
+ original_execute = conn.execute
+ if original_execute.__name__ != 'monkeypatched_execute':
+ def monkeypatched_execute(statement, *multiparams, **params):
+ if isinstance(statement, Select):
+ boundparams = dict(multiparams[0])
+ boundparams.setdefault('_default_language_id', self.default_language_id)
+ multiparams = [boundparams] + list(multiparams[1:])
+ return original_execute(statement, *multiparams, **params)
+ conn.execute = monkeypatched_execute
+ return conn
class MultilangScopedSession(ScopedSession):
"""Dispatches language selection to the attached Session."""
@property
- def default_language(self):
- return self.registry().default_language
-
- @default_language.setter
- def default_language(self, new):
- self.registry().default_language = new
-
- def remove(self):
- del self.registry().default_language
- super(MultilangScopedSession, self).remove()
+ def default_language_id(self):
+ """Passes the new default language id through to the current session.
+ """
+ return self.registry().default_language_id
+
+ @default_language_id.setter
+ def default_language_id(self, new):
+ self.registry().default_language_id = new