Attached is your script worked into separate tests. For test_five,
we can do it like this:
s.query(Document).filter(Document.translations.any(DocumentTranslation.body
== 'doc1_body_en')).all()
For the tests which state "# XXX the current_translation isn't
joinedloaded()", it looks like they are, I added assertions that the
object is present without SQL being emitted; if I remove lazy="joined"
then it fails, so that joinedload is working.
For the other general issue that a query with
Document/DocumentTranslation pulls in other tables, that's because the
relationship has no idea that Document only refers to
DocumentTranslation objects and nothing else. We can make these
relationships explicit. That's in the second attachment. For
test_three, the queries only involve the content/document tables and
there are no LEFT OUTER joins.
For the content_type_id thing, I guess the trigger is so that you can
use raw SQL? SQLAlchemy will do those columns for you within the
ORM. But you can also set polymorphic_on to be a correlated
subquery. the third attachment shows that as well.
In all three cases no changes were needed at all to the persistence side.
The tests include a setup_class() which I don't know how to do in
unittest so I run these with py.test:
py.test test_thing_3.py -k test_three -s
On Thu, Apr 26, 2018 at 6:54 AM, Julien Cigar <[email protected]> wrote:
> Hello Mike,
>
> I finally had the time to make a little full POC, see attached file.
>
> As said previously I'm used to the old mapper() and never used the
> declarative until now, so forgive me if I missed something in the
> script..
>
> Some explanations on the context and on what I'd like to do:
>
> It's a $work-made CMS that is used in some projects here at $work.
> It is developed with the Pyramid framework which allow us to split
> in a very simple and clean way "subprojects" that use our core CMS
> package.
>
> At the core joinedload inheritance is used. "Content" is the base class
> from which other polymorphic entities (Event, Document, Folder, ...)
> inherits.
>
> The discriminator (polymorphic_on) is on a "content_type_id" column in
> the content table which is a foreign key to a content_type table. In my
> real app I use a function to retrieve the content_type.id, which is
> "faked" in my POC (_fake_ids).
>
> Every polymorphic entity has common properties/columns (like
> "title", "description") and specific properties/columns (like "body" for
> a Document, etc). Nothing special, common properties go in the Content
> class and specific properties in their own subclass.
>
> Now I'd like to add a "multilingual layer" to the app. Common properties
> will me moved to a dedicated content_translation table, and each
> polymorphic entity will have their own dedicated table
> (document_translation, event_translation, etc). See attached file
> (sa.png) for the db model. (Note that some entities may not have a
> dedicated _translation table, like Folder, which has no more than a
> "title" and "description")
>
> I'm not sure how to map this in SQLAlchemy as the "translation" parts
> (which are also mapped in a joinedload inheritance) are "part of" or
> "closely linked to" the translatable entity (ex: DocumentTranslation ->
> Document, EventTranslation -> Event, etc).
>
> In an ideal scenario I'd like to have a .translations property
> (relationship()) available in each entity which is lazy='subquery'
> loaded and for which only related _translation tables are JOIN.
> I'm not sure if I should add/overwrite the "translations" relationship
> in every entity (Event, Document, ...)..
>
> In advance, a big "thank you" for the time spent analyzing this :)
>
> Julien
>
> On Fri, Apr 13, 2018 at 10:34:46AM +0200, Julien Cigar wrote:
>> On Thu, Apr 12, 2018 at 08:25:06PM -0400, Mike Bayer wrote:
>> > try sending me a long a full POC and Ill try to play with it, any
>>
>> yep I'll do it ..! I miss some time for now.. but it's definitively on
>> my todo list. I don't think I have an overly complicated use case, it's
>> just that I pay a lot of attention on the generated SQL queries.
>>
>> > reason you arne't using declarative? the classical mappings are hard
>> > to work with.
>>
>> mostly for historical reasons, but also because I prefer the classical
>> mapping approach (mostly for separation of concerns).
>>
>> My application (internal CMS) is based on the Pyramid framework and it's
>> easier to extend the "core" package when the "mapping" part is separated
>> (maybe it's also possible with the declarative approach, I must admit
>> that I haven't looked at it in details).
>>
>> >
>> > On Wed, Apr 11, 2018 at 8:32 AM, Mike Bayer <[email protected]>
>> > wrote:
>> > > On Wed, Apr 11, 2018 at 6:15 AM, Julien Cigar <[email protected]>
>> > > wrote:
>> > >> On Tue, Apr 10, 2018 at 11:53:09AM -0400, Mike Bayer wrote:
>> > >>> On Tue, Apr 10, 2018 at 9:28 AM, Julien Cigar <[email protected]>
>> > >>> wrote:
>> > >>> > Hello,
>> > >>> >
>> > >>> > I wondered if it is possible to use a class mapped against multiple
>> > >>> > tables as a relationship() in another class?
>> > >>>
>> > >>> it is, there's examples at
>> > >>> http://docs.sqlalchemy.org/en/latest/orm/join_conditions.html#relationship-to-non-primary-mapper
>> > >>>
>> > >>> that is map your related class to whatever you want in a non primary
>> > >>> mapper. If you are just selecting data you have a lot of options.
>> > >>>
>> > >>
>> > >> Hi Mike,
>> > >>
>> > >> Thanks for the link :) I tried and it seems to work, except in an
>> > >> inheritance scenario (1). I guess that in a non primary mapper I can't
>> > >> put my polymorphic_on condition on a Join clause?
>> > >>
>> > >> (1) https://gist.github.com/silenius/a237baf8c4bcd79550dc884f2eeb1998
>> > >
>> > > this is a lot of code to follow, but if the point of the mutli-table
>> > > thing you're doing is to game the joined inheritance into doing
>> > > something, then that's probably not the appropriate use for it. It
>> > > looks like you're trying to add all kinds of columns to
>> > > "ContentTranslation" from the other mappers like Content.
>> > >
>> > >
>> > >>
>> > >> Thanks!
>> > >> Julien
>> > >>
>> > >>>
>> > >>> >
>> > >>> > something like:
>> > >>> >
>> > >>> > ### Content
>> > >>> >
>> > >>> > content_translation_join = sql.join(
>> > >>> > t_content, t_content_translation
>> > >>> > )
>> > >>> >
>> > >>> > orm.mapper(
>> > >>> > ContentTranslation, content_translation_join,
>> > >>> > polymorphic_on=t_content.c.content_type_id
>> > >>> > )
>> > >>> >
>> > >>> > orm.mapper(
>> > >>> > Content, t_content,
>> > >>> > polymorphic_on=t_content.c.content_type_id,
>> > >>> > properties={
>> > >>> > 'translations': orm.relationship(
>> > >>> > ContentTranslation,
>> > >>> > # more stuff here
>> > >>> > )
>> > >>> > }
>> > >>> > )
>> > >>> >
>> > >>> > ### Document
>> > >>> >
>> > >>> > orm.mapper(
>> > >>> > DocumentTranslation, t_document_translation,
>> > >>> > inherits=ContentTranslation,
>> > >>> > polymorphic_identity=some_id
>> > >>> > )
>> > >>> >
>> > >>> > orm.mapper(
>> > >>> > Document, t_document,
>> > >>> > inherits=Content,
>> > >>> > polymorphic_identity=some_id
>> > >>> > )
>> > >>> >
>> > >>> > I tried (1) but it doesn't seems to work so I want to be sure that
>> > >>> > it's
>> > >>> > not possible :)
>> > >>> >
>> > >>> > Thanks!
>> > >>> > Julien
>> > >>> >
>> > >>> > (1)
>> > >>> > https://gist.github.com/silenius/561b13f4b987c36434cd81e2c08cab6e#file-foo-py
>> > >>> >
>> > >>> >
>> > >>> >
>> > >>> > --
>> > >>> > Julien Cigar
>> > >>> > Belgian Biodiversity Platform (http://www.biodiversity.be)
>> > >>> > PGP fingerprint: EEF9 F697 4B68 D275 7B11 6A25 B2BB 3710 A204 23C0
>> > >>> > No trees were killed in the creation of this message.
>> > >>> > However, many electrons were terribly inconvenienced.
>> > >>> >
>> > >>> > --
>> > >>> > SQLAlchemy -
>> > >>> > The Python SQL Toolkit and Object Relational Mapper
>> > >>> >
>> > >>> > http://www.sqlalchemy.org/
>> > >>> >
>> > >>> > To post example code, please provide an MCVE: Minimal, Complete, and
>> > >>> > Verifiable Example. See http://stackoverflow.com/help/mcve for a
>> > >>> > full description.
>> > >>> > ---
>> > >>> > You received this message because you are subscribed to the Google
>> > >>> > Groups "sqlalchemy" group.
>> > >>> > To unsubscribe from this group and stop receiving emails from it,
>> > >>> > send an email to [email protected].
>> > >>> > To post to this group, send email to [email protected].
>> > >>> > Visit this group at https://groups.google.com/group/sqlalchemy.
>> > >>> > For more options, visit https://groups.google.com/d/optout.
>> > >>>
>> > >>> --
>> > >>> SQLAlchemy -
>> > >>> The Python SQL Toolkit and Object Relational Mapper
>> > >>>
>> > >>> http://www.sqlalchemy.org/
>> > >>>
>> > >>> To post example code, please provide an MCVE: Minimal, Complete, and
>> > >>> Verifiable Example. See http://stackoverflow.com/help/mcve for a
>> > >>> full description.
>> > >>> ---
>> > >>> You received this message because you are subscribed to the Google
>> > >>> Groups "sqlalchemy" group.
>> > >>> To unsubscribe from this group and stop receiving emails from it, send
>> > >>> an email to [email protected].
>> > >>> To post to this group, send email to [email protected].
>> > >>> Visit this group at https://groups.google.com/group/sqlalchemy.
>> > >>> For more options, visit https://groups.google.com/d/optout.
>> > >>
>> > >> --
>> > >> Julien Cigar
>> > >> Belgian Biodiversity Platform (http://www.biodiversity.be)
>> > >> PGP fingerprint: EEF9 F697 4B68 D275 7B11 6A25 B2BB 3710 A204 23C0
>> > >> No trees were killed in the creation of this message.
>> > >> However, many electrons were terribly inconvenienced.
>> > >>
>> > >> --
>> > >> SQLAlchemy -
>> > >> The Python SQL Toolkit and Object Relational Mapper
>> > >>
>> > >> http://www.sqlalchemy.org/
>> > >>
>> > >> To post example code, please provide an MCVE: Minimal, Complete, and
>> > >> Verifiable Example. See http://stackoverflow.com/help/mcve for a full
>> > >> description.
>> > >> ---
>> > >> You received this message because you are subscribed to the Google
>> > >> Groups "sqlalchemy" group.
>> > >> To unsubscribe from this group and stop receiving emails from it, send
>> > >> an email to [email protected].
>> > >> To post to this group, send email to [email protected].
>> > >> Visit this group at https://groups.google.com/group/sqlalchemy.
>> > >> For more options, visit https://groups.google.com/d/optout.
>> >
>> > --
>> > SQLAlchemy -
>> > The Python SQL Toolkit and Object Relational Mapper
>> >
>> > http://www.sqlalchemy.org/
>> >
>> > To post example code, please provide an MCVE: Minimal, Complete, and
>> > Verifiable Example. See http://stackoverflow.com/help/mcve for a full
>> > description.
>> > ---
>> > You received this message because you are subscribed to the Google Groups
>> > "sqlalchemy" group.
>> > To unsubscribe from this group and stop receiving emails from it, send an
>> > email to [email protected].
>> > To post to this group, send email to [email protected].
>> > Visit this group at https://groups.google.com/group/sqlalchemy.
>> > For more options, visit https://groups.google.com/d/optout.
>>
>> --
>> Julien Cigar
>> Belgian Biodiversity Platform (http://www.biodiversity.be)
>> PGP fingerprint: EEF9 F697 4B68 D275 7B11 6A25 B2BB 3710 A204 23C0
>> No trees were killed in the creation of this message.
>> However, many electrons were terribly inconvenienced.
>>
>> --
>> SQLAlchemy -
>> The Python SQL Toolkit and Object Relational Mapper
>>
>> http://www.sqlalchemy.org/
>>
>> To post example code, please provide an MCVE: Minimal, Complete, and
>> Verifiable Example. See http://stackoverflow.com/help/mcve for a full
>> description.
>> ---
>> You received this message because you are subscribed to the Google Groups
>> "sqlalchemy" group.
>> To unsubscribe from this group and stop receiving emails from it, send an
>> email to [email protected].
>> To post to this group, send email to [email protected].
>> Visit this group at https://groups.google.com/group/sqlalchemy.
>> For more options, visit https://groups.google.com/d/optout.
>
>
>
> --
> Julien Cigar
> Belgian Biodiversity Platform (http://www.biodiversity.be)
> PGP fingerprint: EEF9 F697 4B68 D275 7B11 6A25 B2BB 3710 A204 23C0
> No trees were killed in the creation of this message.
> However, many electrons were terribly inconvenienced.
>
> --
> SQLAlchemy -
> The Python SQL Toolkit and Object Relational Mapper
>
> http://www.sqlalchemy.org/
>
> To post example code, please provide an MCVE: Minimal, Complete, and
> Verifiable Example. See http://stackoverflow.com/help/mcve for a full
> description.
> ---
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to [email protected].
> To post to this group, send email to [email protected].
> Visit this group at https://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/d/optout.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
import logging
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.hybrid import hybrid_property
import unittest
Base = declarative_base()
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
_fake_ids = {
'document': 1,
'event': 2,
'folder': 3
}
def _get_locale():
return 'en'
class ContentType(Base):
__tablename__ = 'content_type'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, id, name):
self.id = id
self.name = name
class Language(Base):
__tablename__ = 'language'
id = Column(String(2), primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, id, name):
self.id = id
self.name = name
class Country(Base):
__tablename__ = 'country'
iso = Column(String(2), primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, iso, name):
self.iso = iso
self.name = name
class Content(Base):
__tablename__ = 'content'
# XXX: not supported by SQLite
#__table_args__ = (
# Index(
# 'idx_container_id',
# text('((1)) where container_id is NULL')
# ),
#)
id = Column(Integer, primary_key=True)
content_type_id = Column(Integer, ForeignKey('content_type.id'),
nullable=False)
container_id = Column(Integer, ForeignKey('folder.content_id'))
parent = relationship(
lambda: Folder,
foreign_keys=lambda: Content.container_id,
innerjoin=True,
uselist=False,
backref=backref('children', cascade='all, delete-orphan')
)
current_translation = relationship(
lambda: ContentTranslation,
primaryjoin=lambda: and_(
ContentTranslation.content_id == Content.id,
ContentTranslation.language_id == bindparam(
None,
callable_=lambda: _get_locale(),
type_=String()
)
),
lazy='joined',
uselist=False,
innerjoin=True,
viewonly=True,
bake_queries=False,
back_populates='content'
)
translations = relationship(
lambda: ContentTranslation,
cascade='all, delete-orphan',
lazy='subquery',
innerjoin=True,
back_populates='content',
collection_class=attribute_mapped_collection('language_id')
)
content_type = relationship(
ContentType,
lazy='joined',
innerjoin=True,
uselist=False
)
__mapper_args__ = {
'polymorphic_on': content_type_id
}
@hybrid_property
def title(self):
return self.current_translation.title
@title.setter
def title(self, value):
self.current_translation.title = value
@hybrid_property
def description(self):
return self.current_translation.description
@description.setter
def description(self, value):
self.current_translation.description = value
class ContentTranslation(Base):
__tablename__ = 'content_translation'
language_id = Column(String(2), ForeignKey('language.id'),
primary_key=True)
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
title = Column(String, nullable=False)
description = Column(String)
# XXX: In an ideal scenario I'd like to put the discriminator on the
# "content.content_type_id" column, something like;
# polymorphic_on = sql.join(
# content_translation, content,
# content_translation.c.content_id == content.c.id
# ).c.content_content_type_id
# But as this is not supported by SQLAlchemy (yet?) I "duplicated" the
# "content_type_id" column and created a trigger which simply select the
# "content_type_id" in the "content" table for the corresponding
# "content_translation.content_id". So in the "real" app there is something
# like:
# create or replace function content_translation_content_id()
# returns trigger as $content_id$
# begin
# NEW.content_type_id := (
# select
# content_type_id
# from
# content
# where
# content.id = NEW.content_id
# );
#
# return NEW;
# end;
# $content_id$ language plpgsql;
# create trigger content_translation_content_id
# before insert or update on content_translation
# for each row execute procedure content_translation_content_id();
content_type_id = Column(Integer, ForeignKey('content_type.id'),
nullable=False)
content = relationship(
Content,
back_populates='translations',
innerjoin=True,
uselist=False
)
content_type = relationship(
ContentType,
lazy='joined',
innerjoin=True,
uselist=False
)
__mapper_args__ = {
'polymorphic_on': content_type_id
}
def __init__(self, **kwargs):
if 'language_id' not in kwargs:
self.language_id = _get_locale()
for k, v in kwargs.items():
setattr(self, k, v)
class Folder(Content):
__tablename__ = 'folder'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
some_folder_col = Column(Integer, nullable=True)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['folder'],
'inherit_condition': content_id == Content.id
}
def __init__(self, title, description=None, some_folder_col=None):
self.some_folder_col = some_folder_col
locale = _get_locale()
self.translations[locale] = FolderTranslation(
title=title, content_type_id=_fake_ids['folder'],
description=description
)
class FolderTranslation(ContentTranslation):
__mapper_args__ = {
'polymorphic_identity': _fake_ids['folder'],
}
class Document(Content):
__tablename__ = 'document'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['document'],
}
def __init__(self, title, body, description=None):
locale = _get_locale()
self.translations[locale] = DocumentTranslation(
body=body, title=title, content_type_id=_fake_ids['document'],
description=description
)
@hybrid_property
def body(self):
return self.current_translation.body
@body.setter
def body(self, value):
self.current_translation.body = value
class DocumentTranslation(ContentTranslation):
__tablename__ = 'document_translation'
language_id = Column(String(2), primary_key=True)
content_id = Column(Integer, ForeignKey('document.content_id'),
primary_key=True)
body = Column(String, nullable=False)
__table_args__ = (
ForeignKeyConstraint(
['language_id', 'content_id'],
['content_translation.language_id',
'content_translation.content_id']
),
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['document'],
'polymorphic_load': 'inline'
}
class Event(Content):
__tablename__ = 'event'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
country_iso = Column(String(2), ForeignKey('country.iso'), nullable=False)
country = relationship(
Country,
innerjoin=True,
uselist=False,
backref='events',
lazy='joined'
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['event'],
}
def __init__(self, title, country_iso, description=None,
some_event_col1=None,
some_event_col2=None):
locale = _get_locale()
self.country_iso = country_iso
self.translations[locale] = EventTranslation(
title=title, content_type_id=_fake_ids['event'],
some_event_col1=some_event_col1, some_event_col2=some_event_col2
)
class EventTranslation(ContentTranslation):
__tablename__ = 'event_translation'
language_id = Column(String(2), primary_key=True)
content_id = Column(Integer, ForeignKey('event.content_id'),
primary_key=True)
some_event_col1 = Column(String())
some_event_col2 = Column(String())
__table_args__ = (
ForeignKeyConstraint(
['language_id', 'content_id'],
['content_translation.language_id',
'content_translation.content_id']
),
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['event'],
'polymorphic_load': 'inline'
}
e = create_engine("sqlite://", echo=False)
Base.metadata.create_all(e)
class FooTest(unittest.TestCase):
@classmethod
def setup_class(cls):
s = Session(e)
for name, fake_id in _fake_ids.items():
s.add(ContentType(fake_id, name))
s.add(Country('be', 'Belgium'))
s.add(Country('fr', 'France'))
s.add(Language('fr', 'Français'))
s.add(Language('en', 'English'))
root = Folder('root')
s.add(root)
subroot = Folder('subroot')
root.children.append(subroot)
doc1 = Document('doc1_title_en', 'doc1_body_en')
doc2 = Document('doc2_title_en', 'doc2_body_en')
subroot.children.append(doc1)
subroot.children.append(doc2)
s.flush()
cls.doc1_id = doc1.id
trans_fr = DocumentTranslation(
language_id='fr',
title='doc1_title_fr',
description='doc1_description_fr',
body='doc1_body_fr'
)
doc1.translations['fr'] = trans_fr
evt1 = Event('Event1', 'be', some_event_col1='sometext')
evt2 = Event('Event2', 'be', some_event_col1='sometext',
some_event_col2="blah")
evt3 = Event('Event3', 'fr', some_event_col2='sometext')
subroot.children.append(evt1)
subroot.children.append(evt2)
subroot.children.append(evt3)
s.commit()
def test_one(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
s.close()
assert foo.current_translation
def test_two(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
s.close()
assert foo.current_translation
assert foo.translations
def test_three(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
# .body is specific to document_translation
# without polymorphic_load='inline' a query is made .. in practice
# with the lazy='subquery' on the "translations" relationship it
# should be the case however, with polymorphic_load='inline' the
#table is always joined, even for non-Document entities
s.close()
assert foo.current_translation
assert foo.translations
foo.translations['en'].body
def test_four(self):
s = Session(e)
s.query(Document).filter(Document.translations.any(language_id='fr')).all()
def test_five(self):
s = Session(e)
# XXX should work, but fail, as .body is specific to DocumentTranslation
s.query(Document).filter(Document.translations.any(DocumentTranslation.body == 'doc1_body_en')).all()
import logging
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.hybrid import hybrid_property
import unittest
Base = declarative_base()
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
_fake_ids = {
'document': 1,
'event': 2,
'folder': 3
}
def _get_locale():
return 'en'
class ContentType(Base):
__tablename__ = 'content_type'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, id, name):
self.id = id
self.name = name
class Language(Base):
__tablename__ = 'language'
id = Column(String(2), primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, id, name):
self.id = id
self.name = name
class Country(Base):
__tablename__ = 'country'
iso = Column(String(2), primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, iso, name):
self.iso = iso
self.name = name
def _setup_translation(content_cls, translation_cls):
content_cls.current_translation = relationship(
lambda: translation_cls,
primaryjoin=lambda: and_(
foreign(translation_cls.content_id) == Content.id,
translation_cls.language_id == bindparam(
None,
callable_=lambda: _get_locale(),
type_=String()
)
),
lazy='joined',
uselist=False,
innerjoin=True,
viewonly=True,
bake_queries=False,
back_populates='content'
)
content_cls.translations = relationship(
lambda: translation_cls,
cascade='all, delete-orphan',
lazy='subquery',
innerjoin=True,
back_populates='content',
collection_class=attribute_mapped_collection('language_id')
)
translation_cls.content = relationship(
content_cls,
back_populates='translations',
innerjoin=True,
uselist=False
)
class Content(Base):
__tablename__ = 'content'
id = Column(Integer, primary_key=True)
content_type_id = Column(Integer, ForeignKey('content_type.id'),
nullable=False)
container_id = Column(Integer, ForeignKey('folder.content_id'))
parent = relationship(
lambda: Folder,
foreign_keys=lambda: Content.container_id,
innerjoin=True,
uselist=False,
backref=backref('children', cascade='all, delete-orphan')
)
content_type = relationship(
ContentType,
lazy='joined',
innerjoin=True,
uselist=False
)
__mapper_args__ = {
'polymorphic_on': content_type_id
}
@hybrid_property
def title(self):
return self.current_translation.title
@title.setter
def title(self, value):
self.current_translation.title = value
@hybrid_property
def description(self):
return self.current_translation.description
@description.setter
def description(self, value):
self.current_translation.description = value
class ContentTranslation(Base):
__tablename__ = 'content_translation'
language_id = Column(String(2), ForeignKey('language.id'),
primary_key=True)
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
title = Column(String, nullable=False)
description = Column(String)
# XXX: In an ideal scenario I'd like to put the discriminator on the
# "content.content_type_id" column, something like;
# polymorphic_on = sql.join(
# content_translation, content,
# content_translation.c.content_id == content.c.id
# ).c.content_content_type_id
# But as this is not supported by SQLAlchemy (yet?) I "duplicated" the
# "content_type_id" column and created a trigger which simply select the
# "content_type_id" in the "content" table for the corresponding
# "content_translation.content_id". So in the "real" app there is something
# like:
# create or replace function content_translation_content_id()
# returns trigger as $content_id$
# begin
# NEW.content_type_id := (
# select
# content_type_id
# from
# content
# where
# content.id = NEW.content_id
# );
#
# return NEW;
# end;
# $content_id$ language plpgsql;
# create trigger content_translation_content_id
# before insert or update on content_translation
# for each row execute procedure content_translation_content_id();
content_type_id = Column(Integer, ForeignKey('content_type.id'),
nullable=False)
content_type = relationship(
ContentType,
lazy='joined',
innerjoin=True,
uselist=False
)
__mapper_args__ = {
'polymorphic_on': content_type_id
}
def __init__(self, **kwargs):
if 'language_id' not in kwargs:
self.language_id = _get_locale()
for k, v in kwargs.items():
setattr(self, k, v)
class Folder(Content):
__tablename__ = 'folder'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
some_folder_col = Column(Integer, nullable=True)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['folder'],
'inherit_condition': content_id == Content.id
}
def __init__(self, title, description=None, some_folder_col=None):
self.some_folder_col = some_folder_col
locale = _get_locale()
self.translations[locale] = FolderTranslation(
title=title, content_type_id=_fake_ids['folder'],
description=description
)
class FolderTranslation(ContentTranslation):
__mapper_args__ = {
'polymorphic_identity': _fake_ids['folder'],
}
class Document(Content):
__tablename__ = 'document'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['document'],
}
def __init__(self, title, body, description=None):
locale = _get_locale()
self.translations[locale] = DocumentTranslation(
body=body, title=title, content_type_id=_fake_ids['document'],
description=description
)
@hybrid_property
def body(self):
return self.current_translation.body
@body.setter
def body(self, value):
self.current_translation.body = value
class DocumentTranslation(ContentTranslation):
__tablename__ = 'document_translation'
language_id = Column(String(2), primary_key=True)
content_id = Column(Integer, ForeignKey('document.content_id'),
primary_key=True)
body = Column(String, nullable=False)
__table_args__ = (
ForeignKeyConstraint(
['language_id', 'content_id'],
['content_translation.language_id',
'content_translation.content_id']
),
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['document'],
'polymorphic_load': 'inline'
}
class Event(Content):
__tablename__ = 'event'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
country_iso = Column(String(2), ForeignKey('country.iso'), nullable=False)
country = relationship(
Country,
innerjoin=True,
uselist=False,
backref='events',
lazy='joined'
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['event'],
}
def __init__(self, title, country_iso, description=None,
some_event_col1=None,
some_event_col2=None):
locale = _get_locale()
self.country_iso = country_iso
self.translations[locale] = EventTranslation(
title=title, content_type_id=_fake_ids['event'],
some_event_col1=some_event_col1, some_event_col2=some_event_col2
)
class EventTranslation(ContentTranslation):
__tablename__ = 'event_translation'
language_id = Column(String(2), primary_key=True)
content_id = Column(Integer, ForeignKey('event.content_id'),
primary_key=True)
some_event_col1 = Column(String())
some_event_col2 = Column(String())
__table_args__ = (
ForeignKeyConstraint(
['language_id', 'content_id'],
['content_translation.language_id',
'content_translation.content_id']
),
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['event'],
'polymorphic_load': 'inline'
}
_setup_translation(Content, ContentTranslation)
_setup_translation(Folder, FolderTranslation)
_setup_translation(Document, DocumentTranslation)
_setup_translation(Event, EventTranslation)
e = create_engine("sqlite://", echo=False)
Base.metadata.create_all(e)
class FooTest(unittest.TestCase):
@classmethod
def setup_class(cls):
s = Session(e)
for name, fake_id in _fake_ids.items():
s.add(ContentType(fake_id, name))
s.add(Country('be', 'Belgium'))
s.add(Country('fr', 'France'))
s.flush()
s.add(Language('fr', 'Français'))
s.add(Language('en', 'English'))
s.flush()
root = Folder('root')
s.add(root)
s.flush()
subroot = Folder('subroot')
root.children.append(subroot)
s.flush()
doc1 = Document('doc1_title_en', 'doc1_body_en')
doc2 = Document('doc2_title_en', 'doc2_body_en')
subroot.children.append(doc1)
subroot.children.append(doc2)
s.flush()
cls.doc1_id = doc1.id
trans_fr = DocumentTranslation(
language_id='fr',
title='doc1_title_fr',
description='doc1_description_fr',
body='doc1_body_fr'
)
doc1.translations['fr'] = trans_fr
s.flush()
evt1 = Event('Event1', 'be', some_event_col1='sometext')
evt2 = Event('Event2', 'be', some_event_col1='sometext',
some_event_col2="blah")
evt3 = Event('Event3', 'fr', some_event_col2='sometext')
subroot.children.append(evt1)
subroot.children.append(evt2)
subroot.children.append(evt3)
s.commit()
def test_one(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
s.close()
assert foo.current_translation
def test_two(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
foo.translations
def test_three(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
# .body is specific to document_translation
# without polymorphic_load='inline' a query is made .. in practice
# with the lazy='subquery' on the "translations" relationship it
# should be the case however, with polymorphic_load='inline' the
#table is always joined, even for non-Document entities
foo.translations['en'].body
def test_four(self):
s = Session(e)
s.query(Document).filter(Document.translations.any(language_id='fr')).all()
def test_five(self):
s = Session(e)
# XXX should work, but fail, as .body is specific to DocumentTranslation
s.query(Document).filter(Document.translations.any(body='doc1_body_en')).all()
import logging
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.hybrid import hybrid_property
import unittest
Base = declarative_base()
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
_fake_ids = {
'document': 1,
'event': 2,
'folder': 3
}
def _get_locale():
return 'en'
class ContentType(Base):
__tablename__ = 'content_type'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, id, name):
self.id = id
self.name = name
class Language(Base):
__tablename__ = 'language'
id = Column(String(2), primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, id, name):
self.id = id
self.name = name
class Country(Base):
__tablename__ = 'country'
iso = Column(String(2), primary_key=True)
name = Column(String(50), nullable=False)
def __init__(self, iso, name):
self.iso = iso
self.name = name
def _setup_translation(content_cls, translation_cls):
content_cls.current_translation = relationship(
lambda: translation_cls,
primaryjoin=lambda: and_(
foreign(translation_cls.content_id) == Content.id,
translation_cls.language_id == bindparam(
None,
callable_=lambda: _get_locale(),
type_=String()
)
),
lazy='joined',
uselist=False,
innerjoin=True,
viewonly=True,
bake_queries=False,
back_populates='content'
)
content_cls.translations = relationship(
lambda: translation_cls,
cascade='all, delete-orphan',
lazy='subquery',
innerjoin=True,
back_populates='content',
collection_class=attribute_mapped_collection('language_id')
)
translation_cls.content = relationship(
content_cls,
back_populates='translations',
innerjoin=True,
uselist=False
)
class Content(Base):
__tablename__ = 'content'
id = Column(Integer, primary_key=True)
content_type_id = Column(Integer, ForeignKey('content_type.id'),
nullable=False)
container_id = Column(Integer, ForeignKey('folder.content_id'))
parent = relationship(
lambda: Folder,
foreign_keys=lambda: Content.container_id,
innerjoin=True,
uselist=False,
backref=backref('children', cascade='all, delete-orphan')
)
content_type = relationship(
ContentType,
lazy='joined',
innerjoin=True,
uselist=False
)
__mapper_args__ = {
'polymorphic_on': content_type_id
}
@hybrid_property
def title(self):
return self.current_translation.title
@title.setter
def title(self, value):
self.current_translation.title = value
@hybrid_property
def description(self):
return self.current_translation.description
@description.setter
def description(self, value):
self.current_translation.description = value
class ContentTranslation(Base):
__tablename__ = 'content_translation'
language_id = Column(String(2), ForeignKey('language.id'),
primary_key=True)
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
title = Column(String, nullable=False)
description = Column(String)
content_type = association_proxy("content", "content_type")
__mapper_args__ = {
'polymorphic_on':
select([Content.content_type_id]).where(content_id == Content.id).\
correlate_except(Content).as_scalar()
}
def __init__(self, **kwargs):
if 'language_id' not in kwargs:
self.language_id = _get_locale()
for k, v in kwargs.items():
setattr(self, k, v)
class Folder(Content):
__tablename__ = 'folder'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
some_folder_col = Column(Integer, nullable=True)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['folder'],
'inherit_condition': content_id == Content.id
}
def __init__(self, title, description=None, some_folder_col=None):
self.some_folder_col = some_folder_col
locale = _get_locale()
self.translations[locale] = FolderTranslation(
title=title, content_type_id=_fake_ids['folder'],
description=description
)
class FolderTranslation(ContentTranslation):
__mapper_args__ = {
'polymorphic_identity': _fake_ids['folder'],
}
class Document(Content):
__tablename__ = 'document'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['document'],
}
def __init__(self, title, body, description=None):
locale = _get_locale()
self.translations[locale] = DocumentTranslation(
body=body, title=title, content_type_id=_fake_ids['document'],
description=description
)
@hybrid_property
def body(self):
return self.current_translation.body
@body.setter
def body(self, value):
self.current_translation.body = value
class DocumentTranslation(ContentTranslation):
__tablename__ = 'document_translation'
language_id = Column(String(2), primary_key=True)
content_id = Column(Integer, ForeignKey('document.content_id'),
primary_key=True)
body = Column(String, nullable=False)
__table_args__ = (
ForeignKeyConstraint(
['language_id', 'content_id'],
['content_translation.language_id',
'content_translation.content_id']
),
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['document'],
'polymorphic_load': 'inline'
}
class Event(Content):
__tablename__ = 'event'
content_id = Column(Integer, ForeignKey('content.id'), primary_key=True)
country_iso = Column(String(2), ForeignKey('country.iso'), nullable=False)
country = relationship(
Country,
innerjoin=True,
uselist=False,
backref='events',
lazy='joined'
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['event'],
}
def __init__(self, title, country_iso, description=None,
some_event_col1=None,
some_event_col2=None):
locale = _get_locale()
self.country_iso = country_iso
self.translations[locale] = EventTranslation(
title=title, content_type_id=_fake_ids['event'],
some_event_col1=some_event_col1, some_event_col2=some_event_col2
)
class EventTranslation(ContentTranslation):
__tablename__ = 'event_translation'
language_id = Column(String(2), primary_key=True)
content_id = Column(Integer, ForeignKey('event.content_id'),
primary_key=True)
some_event_col1 = Column(String())
some_event_col2 = Column(String())
__table_args__ = (
ForeignKeyConstraint(
['language_id', 'content_id'],
['content_translation.language_id',
'content_translation.content_id']
),
)
__mapper_args__ = {
'polymorphic_identity': _fake_ids['event'],
'polymorphic_load': 'inline'
}
_setup_translation(Content, ContentTranslation)
_setup_translation(Folder, FolderTranslation)
_setup_translation(Document, DocumentTranslation)
_setup_translation(Event, EventTranslation)
e = create_engine("sqlite://", echo=False)
Base.metadata.create_all(e)
class FooTest(unittest.TestCase):
@classmethod
def setup_class(cls):
s = Session(e)
for name, fake_id in _fake_ids.items():
s.add(ContentType(fake_id, name))
s.add(Country('be', 'Belgium'))
s.add(Country('fr', 'France'))
s.flush()
s.add(Language('fr', 'Français'))
s.add(Language('en', 'English'))
s.flush()
root = Folder('root')
s.add(root)
s.flush()
subroot = Folder('subroot')
root.children.append(subroot)
s.flush()
doc1 = Document('doc1_title_en', 'doc1_body_en')
doc2 = Document('doc2_title_en', 'doc2_body_en')
subroot.children.append(doc1)
subroot.children.append(doc2)
s.flush()
cls.doc1_id = doc1.id
trans_fr = DocumentTranslation(
language_id='fr',
title='doc1_title_fr',
description='doc1_description_fr',
body='doc1_body_fr'
)
doc1.translations['fr'] = trans_fr
s.flush()
evt1 = Event('Event1', 'be', some_event_col1='sometext')
evt2 = Event('Event2', 'be', some_event_col1='sometext',
some_event_col2="blah")
evt3 = Event('Event3', 'fr', some_event_col2='sometext')
subroot.children.append(evt1)
subroot.children.append(evt2)
subroot.children.append(evt3)
s.commit()
def test_one(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
s.close()
assert foo.current_translation
def test_two(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
foo.translations
def test_three(self):
s = Session(e)
foo = s.query(Document).get(self.doc1_id) # XXX the current_translation isn't joinedloaded()
# .body is specific to document_translation
# without polymorphic_load='inline' a query is made .. in practice
# with the lazy='subquery' on the "translations" relationship it
# should be the case however, with polymorphic_load='inline' the
#table is always joined, even for non-Document entities
foo.translations['en'].body
def test_four(self):
s = Session(e)
s.query(Document).filter(Document.translations.any(language_id='fr')).all()
def test_five(self):
s = Session(e)
# XXX should work, but fail, as .body is specific to DocumentTranslation
s.query(Document).filter(Document.translations.any(body='doc1_body_en')).all()
