sa_assoc_test.py

"""
demonstration of an obscure edge case / bug(?) in sqlalchemy where
adding a from_self() to a query that's already been aliased once and
contains a labeled literal will fail.

summary
1 - from_self
2 - add a non-column entity like count()
3 - another from_self fails

the bug is introduced in r6402, which has the commit message:

  - query.from_self(), query.union(), others which do a 
    "SELECT * from (SELECT...)" type of nesting will do
    a better job translating column expressions within the subquery
    to the columns clause of the outer query.  This is
    potentially backwards incompatible with 0.5, in that this 
    may break queries with literal expressions that do not have labels
    applied (i.e. literal('foo'), etc.)
    [ticket:1568]
"""

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy

metadata = MetaData('sqlite://')
Base = declarative_base(metadata=metadata)

class Foo(Base):
    __tablename__ = 'foos'
    id = Column(Integer, primary_key=True)
    text = Column(String(50))
    def __init__(self, text):
        self.text = text
    assocs = relation('Assoc')
    bars = association_proxy('assocs', 'bar')

class Assoc(Base):
    __tablename__ = 'assoc'
    foo_id = Column(None, ForeignKey('foos.id'), primary_key=True)
    bar_id = Column(None, ForeignKey('bars.id'), primary_key=True)
    bar = relation('Bar')
    def __init__(self, bar):
        self.bar = bar

class Bar(Base):
    __tablename__ = 'bars'
    id = Column(Integer, primary_key=True)
    text = Column(String(50))
    def __init__(self, text):
        self.text = text

metadata.drop_all()
metadata.create_all()

foo1 = Foo('foo1')
bar1 = Bar('bar1')
bar2 = Bar('bar2')
foo1.bars.append(bar1)
foo1.bars.append(bar2)

sess = create_session()
sess.add_all([foo1, bar1, bar2])

sess.flush()
sess.expunge_all()

q = sess.query(Foo).filter(Foo.text.contains('o'))
print q.all()
print q.from_self().\
        join(Foo.assocs).\
        join(Assoc.bar).\
        add_entity(Bar).\
        group_by(Bar.id).\
        add_column(func.count().label('num')).\
        from_self().\
        all()