implement "immutable" argument for ClauseAdapter; apply to aliasing in Query._compile

SQL level issue:

from sqlalchemy import *
from sqlalchemy.orm import *

metadata = MetaData()

elements = Table('elements', metadata,
   Column('element_id', Integer, primary_key=True),
   Column('data', String(30))
)

items = Table('items', metadata,
    Column('item_id', Integer, primary_key=True),
   Column('element_id', Integer, ForeignKey('elements.element_id'), primary_key=True),
   Column('moredata', String(30))
)

items_select = select(
   [elements.c.data](items,),
   items.c.element_id==elements.c.element_id
   ).alias('element_items')


elements_alias = elements.alias()
j = items_select.outerjoin(elements_alias, items_select.c.element_id==elements_alias.c.element_id)

from sqlalchemy.sql.util import ClauseAdapter

s1 = select([items_select](items_select)).limit(5).offset(10).alias()

print "------------------"
print s1
print "------------------"
print j
print "------------------"

print ClauseAdapter(s1).traverse(j, stop_on=set([items_select](items_select)))

print "------------------"

ORM test (i.e. due to limit/offset):

engine = create_engine('sqlite:///', echo=True)

metadata.bind = engine
metadata.create_all()

class Item(object):
   pass
class Element(object):
    pass

mapper(Item, items_select, properties={
    'elements':relation(Element, lazy=False, primaryjoin=items_select.c.element_id==elements.c.element_id)
})
mapper(Element, elements)

print class_mapper(Item).primary_key
assert set(class_mapper(Item).primary_key) == set([items_select.c.item_id](items_select.c.element_id,))

sess = create_session()
sess.query(Item).first()

end-user test case. all of the inheritance, dynamic=True etc. has nothing to do with it. im 90% sure its the same issue but the broken SQL is slightly different:

from sqlalchemy import *
from sqlalchemy.orm import *


metadata = MetaData()


elements = Table('elements', metadata,
   Column('id', Integer, primary_key=True),
   Column('type', String(128)),
   Column('name', String(256)),
)


items = Table('items', metadata,
   Column('assembly_id', Integer, ForeignKey('elements.id'), primary_key=True),
   Column('element_id', Integer, ForeignKey('elements.id'), primary_key=True),
   Column('quantity', Integer, default=1, nullable=False),
)


engine = create_engine('sqlite:///')

metadata.bind = engine
metadata.create_all()


class Element(object):

   def __init__(self, name):
       self.name = name

   def __repr__(self):
       return '<%s %s>' % (self.__class__.__name__, self.name)

class Part(Element): pass

class Assembly(Element): pass

class Product(Assembly): pass


class Item(object):

   def __init__(self, assembly, element, quantity=1):
       self.assembly = assembly
       self.element = element
       self.quantity = quantity

   def __repr__(self):
       return '<%s %d %s %s>' % (self.__class__.__name__, self.quantity,
                              self.assembly.name, self.element.name)


element_mapper = mapper(Element, elements,
   polymorphic_on=elements.c.type,
   polymorphic_identity='element'
)

part_mapper = mapper(Part, inherits=element_mapper, polymorphic_identity='part')
assembly_mapper = mapper(Assembly, inherits=element_mapper, polymorphic_identity='assembly')
product_mapper = mapper(Product, inherits=assembly_mapper, polymorphic_identity='product')


items_select = select(
   [elements.c.name](items,),
   items.c.element_id==elements.c.id
   ).alias('element_items')

items_mapper = mapper(Item, items_select,
   primary_key=[items_select.c.element_id](items_select.c.assembly_id,),
   order_by=items_select.c.name,
   properties=dict(
       assembly=relation(
           Assembly, lazy=False, uselist=False,
           foreign_keys=[items_select.c.assembly_id](items_select.c.assembly_id),
           primaryjoin=items_select.c.assembly_id==elements.c.id,
           backref=backref('element_items', lazy='dynamic',
                           primaryjoin=items_select.c.assembly_id==elements.c.id),
       ),
       element=relation(
           Element, lazy=False, uselist=False,
           foreign_keys=[items_select.c.element_id](items_select.c.element_id),
           primaryjoin=items_select.c.element_id==elements.c.id,
           backref=backref('assembly_items', lazy='dynamic',
                           primaryjoin=items_select.c.element_id==elements.c.id),
       ),
       quantity=items_select.c.quantity,
   )
)


session = create_session(bind=engine)


prod1 = Product('prod1')
session.save(prod1)

part1 = Part('part1')
session.save(part1)

for i in range(2):
   assem = Assembly('assem%d' % i)
   i1 = Item(prod1, assem)
   i2 = Item(assem, part1)
   [for o in (assem, i1, i2)](session.save(o))

session.flush()
session.clear()


prod1 = session.query(Product).filter_by(name='prod1').one()

print prod1.element_items[0](0)

# or just do this.  the LIMIT/OFFEST aliasing is the issue:
session.query(Item).first()

metadata.drop_all()

Comments (3)