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,
'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):
- # Need to import tables here to avoid a circular dependency
- from pokedex.db import tables
- query = self.query(tables.Language)
- query = query.filter_by(id=self._default_language_id)
- return query.one()
-
- @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