Hi guys, I found very strange behaviour of my script, after a lot of debugging it appears that SQLAlchemy complains about joins that I cannot see exactly why it has any issue with.
Full running example is below in both as inline main.py script and link. In the nutshell, hierarchy of objects is as follows. Article has Video, Video has Tag Series, Tag has Slug field Article has as well Tags as m2m, each tag has Slug field. What is wrong? When I do Article.join(ArticleTag).join(Tag).join(SlugTag).join(Slug).join(Video).join(Tag[Series]).join(SlugTag).join(Slug) I will get error that SQLAlchemy cannot join Video as it's ambiguous join. When I Do Article.join(Video).join(Tag[Series]).join(SlugTag).join(Slug).join(ArticleTag).join(Tag).join(SlugTag).join(Slug). All works like a charm. Different solution is to specify that when joining Video, onclause should be Article.video_uid == Video.uid. In general Article can have only one Video and nothing else can do it. So I wonder where SQLAlchemy finds this ambiguity. Is it a bug? Keep in mind this code is a small piece of bigger picture, so you might be wondering why I have this aliasing feature. It's wrapper, that collects all filters/joins and then compiles them removing for example duplicated joins). I removed most of it but there might be an artefact or two. Link: http://sprunge.us/KBeM Inline code: # encoding: utf8 from __future__ import unicode_literals import sys import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Slug(Base): __tablename__ = 'slug' id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) slug = sqlalchemy.Column(sqlalchemy.Unicode(255), nullable=False) expiration_date = sqlalchemy.Column( sqlalchemy.DateTime(timezone=True), default=None) class SlugTag(Base): __tablename__ = 'slug_tag' slug = sqlalchemy.orm.relationship( 'Slug', primaryjoin='SlugTag.slug_id == Slug.id', cascade="all, delete-orphan", uselist=False, single_parent=True) obj = sqlalchemy.orm.relationship( 'Tag', primaryjoin='SlugTag.obj_uid == Tag.uid', uselist=False) slug_id = sqlalchemy.Column( sqlalchemy.Integer, sqlalchemy.ForeignKey('slug.id'), nullable=False) obj_uid = sqlalchemy.Column( sqlalchemy.Unicode(32), sqlalchemy.ForeignKey('tag.uid'), nullable=False) __table_args__ = ( sqlalchemy.PrimaryKeyConstraint( 'slug_id', 'obj_uid', name='slug_tag_pk'), ) class Tag(Base): __tablename__ = 'tag' uid = sqlalchemy.Column( sqlalchemy.Unicode(32), primary_key=True, nullable=False) title = sqlalchemy.Column(sqlalchemy.Unicode(255), nullable=False) type = sqlalchemy.Column(sqlalchemy.Unicode(255)) slugs_connector = sqlalchemy.orm.relationship( 'SlugTag', primaryjoin='Tag.uid == SlugTag.obj_uid') __table_args__ = ( sqlalchemy.UniqueConstraint('type', 'title', name='title_type_unique'), ) class Video(Base): __tablename__ = 'video' SERIES_TYPE = 'series' uid = sqlalchemy.Column( sqlalchemy.Unicode(32), primary_key=True, nullable=False) title = sqlalchemy.Column(sqlalchemy.Unicode(255), nullable=False) body = sqlalchemy.Column(sqlalchemy.Unicode) duration = sqlalchemy.Column(sqlalchemy.Time) series_uid = sqlalchemy.Column( sqlalchemy.Unicode(32), sqlalchemy.ForeignKey('tag.uid'), index=True, nullable=False) series = sqlalchemy.orm.relationship( 'Tag', primaryjoin='Video.series_uid == Tag.uid', uselist=False) class ArticleTag(Base): """Represent m2m table between Tags and Article. association obj for mapping m2m articles<->tags """ __tablename__ = 'article_tag' article_uid = sqlalchemy.Column( sqlalchemy.Unicode(32), sqlalchemy.ForeignKey('article.uid'), primary_key=True) tag_uid = sqlalchemy.Column( sqlalchemy.Unicode(32), sqlalchemy.ForeignKey('tag.uid'), primary_key=True) tag = sqlalchemy.orm.relationship('Tag', backref='articles') class Article(Base): __tablename__ = 'article' uid = sqlalchemy.Column( sqlalchemy.Unicode(32), primary_key=True, nullable=False) title = sqlalchemy.Column(sqlalchemy.Unicode(255), nullable=False) meta_description = sqlalchemy.Column(sqlalchemy.Unicode(255)) meta_title = sqlalchemy.Column(sqlalchemy.Unicode(255)) body = sqlalchemy.Column(sqlalchemy.Unicode) video_uid = sqlalchemy.Column( sqlalchemy.Unicode(32), sqlalchemy.ForeignKey('video.uid'), index=True) video = sqlalchemy.orm.relationship( 'Video', primaryjoin='Article.video_uid == Video.uid', uselist=False, backref='articles') article_tags = sqlalchemy.orm.relationship('ArticleTag', backref='article') def _alias_onclause(join, onclause): if hasattr(onclause, 'clauses'): for clause in onclause.clauses: _alias_onclause(join, clause) return ltable = getattr(onclause.left, 'table', None) rtable = getattr(onclause.right, 'table', None) if ltable is join.__table__: onclause.left = getattr(join, onclause.left.name) if rtable is join.__table__: onclause.right = getattr(join, onclause.right.name) def main(): engine = create_engine('sqlite:///:memory:') Session = sessionmaker(bind=engine)() Base.metadata.create_all(engine) joins = [] # join #A joins.append((ArticleTag, ArticleTag.article_uid == Article.uid)) joins.append((Tag, None)) aslugtag = sqlalchemy.orm.util.AliasedClass(SlugTag) aslug = sqlalchemy.orm.util.AliasedClass(Slug) joins.append((aslugtag, None)) joins.append(( aslug, (aslug.id == aslugtag.slug_id) & (aslug.expiration_date.is_(None)))) # end of join #A # join #B # if success True there is extra onclause success = False if success: joins.append((Video, Video.uid == Article.video_uid)) else: joins.append((Video, None)) joins.append(( Tag, (Video.series_uid == Tag.uid) & (Tag.type == Video.SERIES_TYPE))) aslugtag = sqlalchemy.orm.util.AliasedClass(SlugTag) aslug = sqlalchemy.orm.util.AliasedClass(Slug) joins.append((aslugtag, None)) joins.append(( aslug, (aslug.id == aslugtag.slug_id) & (aslug.expiration_date.is_(None)))) # end of join #B inner_joins = [] qry = Session.query(Article) for join, onclause in joins: if join in inner_joins: join = sqlalchemy.orm.util.AliasedClass(join) if onclause is not None: _alias_onclause(join, onclause) if onclause is None: qry = qry.join(join) else: qry = qry.join(join, onclause) inner_joins.append(join) print qry.count() print('sys', sys.version) print('SQLAlchemy', sqlalchemy.__version__) if __name__ == '__main__': main() -- 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.
