On Thu, Apr 26, 2018 at 11:52:48AM -0400, Mike Bayer wrote: > 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 >
Hi Mike, It works almost like a charm. One minor issue I have is in the test_six I've added in the attached test_thing_3.py. When I'm querying through the base class I'm wondering why SQLAlchemy issues a query for a property which is already lazy='joined' loaded in the previously query, and I wonder if there is way to avoid that .. ? Thanks ! Julien > > > > On Thu, Apr 26, 2018 at 6:54 AM, Julien Cigar <julien.ci...@gmail.com> 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 <mike...@zzzcomputing.com> > >> > wrote: > >> > > On Wed, Apr 11, 2018 at 6:15 AM, Julien Cigar <julien.ci...@gmail.com> > >> > > 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 > >> > >>> <julien.ci...@gmail.com> 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 sqlalchemy+unsubscr...@googlegroups.com. > >> > >>> > To post to this group, send email to sqlalchemy@googlegroups.com. > >> > >>> > 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 sqlalchemy+unsubscr...@googlegroups.com. > >> > >>> To post to this group, send email to sqlalchemy@googlegroups.com. > >> > >>> 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 sqlalchemy+unsubscr...@googlegroups.com. > >> > >> To post to this group, send email to sqlalchemy@googlegroups.com. > >> > >> 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 sqlalchemy+unsubscr...@googlegroups.com. > >> > To post to this group, send email to sqlalchemy@googlegroups.com. > >> > 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 sqlalchemy+unsubscr...@googlegroups.com. > >> To post to this group, send email to sqlalchemy@googlegroups.com. > >> 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 sqlalchemy+unsubscr...@googlegroups.com. > > To post to this group, send email to sqlalchemy@googlegroups.com. > > 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 sqlalchemy+unsubscr...@googlegroups.com. > To post to this group, send email to sqlalchemy@googlegroups.com. > 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() -- 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 sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. 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.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()
def test_six(self):
s = Session(e)
a = s.query(Content).all() # note: current_translation is joinedload
[_.title for _ in a]
signature.asc
Description: PGP signature
