filter(collection.of_type(x).any()) fails to match

from sqlalchemy import *
from sqlalchemy.orm import *


def eq_(a, b):
    assert a == b, "%r != %r" % (a, b)

metadata = MetaData()
Table('principals', metadata,
      Column('id', Integer, primary_key=True),
      Column('principal_type', String(16), nullable=False))

Table('users', metadata,
      Column('id', Integer, ForeignKey('principals.id'), primary_key=True),
      Column('name', String(200), nullable=False),
      Column('creator_id', Integer, ForeignKey('principals.id'),
             nullable=False))

Table('groups', metadata,
      Column('id', Integer, ForeignKey('principals.id'), primary_key=True),
      Column('name', String(200), nullable=False),
      Column('description', String(255)))

Table('group_members', metadata,
      Column('group_id', Integer, ForeignKey('groups.id'),
             nullable=False, primary_key=True),
      Column('principal_id', Integer, ForeignKey('principals.id'),
             nullable=False, primary_key=True))


for name, value in metadata.tables.items():
    locals()[name](name) = value

class Principal(object):
    pass

class Group(Principal):
    pass

class User(Principal):
    pass


mapper(Principal, principals,
       polymorphic_on=principals.c.principal_type,
       properties={
         'parents': relation(Group, secondary=group_members, backref='members'),
       })

mapper(User, users,
       inherits=Principal,
       inherit_condition=principals.c.id == users.c.id,
       polymorphic_identity='User')

mapper(Group, groups,
       inherits=Principal,
       polymorphic_identity='Group')

def data():
    principals.insert().execute(
        {'id': 1, 'principal_type': 'User'},
        {'id': 2, 'principal_type': 'Group'},
        {'id': 3, 'principal_type': 'Group'},
        {'id': 4, 'principal_type': 'User'})
    users.insert().execute(
        {'id': 1, 'name': 'user @ 1', 'creator_id': 1},
        {'id': 4, 'name': 'user @ 4', 'creator_id': 1})
    groups.insert().execute(
        {'id': 2, 'name': 'group @ 2'},
        {'id': 3, 'name': 'group @ 3'})
    group_members.insert().execute(
        {'group_id': 2, 'principal_id': 1},
        {'group_id': 2, 'principal_id': 3},
        {'group_id': 3, 'principal_id': 4})

def test():
    s = create_session()

    eq_(2, s.query(Group).filter(Group.members.of_type(User).any()).count())
    eq_(2, len(s.query(Group).filter(Group.members.of_type(User).any()).all()))

    eq_(1, len(s.query(Group).filter(Group.members.of_type(Group).any()).all()))
    eq_(1, s.query(Group).filter(Group.members.of_type(Group).any()).count())

if __name__ == '__main__':
    metadata.bind = 'sqlite:///'
    metadata.create_all()
    try:
        data()
        test()
    finally:
        metadata.drop_all()
Comments (2)
