from sqlalchemy.test.testing import eq_
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.test import testing
from test.orm import _base
class InheritTest(_base.MappedTest):
"""deals with inheritance and many-to-many relationships"""
@classmethod
def define_tables(cls, metadata):
global principals
global users
global groups
global user_group_map
principals = Table('principals', metadata,
Column('principal_id', Integer,
Sequence('principal_id_seq', optional=False),
primary_key=True),
Column('name', String(50), nullable=False))
users = Table('prin_users', metadata,
Column('principal_id', Integer,
ForeignKey('principals.principal_id'), primary_key=True),
Column('password', String(50), nullable=False),
Column('email', String(50), nullable=False),
Column('login_id', String(50), nullable=False))
groups = Table('prin_groups', metadata,
Column('principal_id', Integer,
ForeignKey('principals.principal_id'), primary_key=True))
user_group_map = Table('prin_user_group_map', metadata,
Column('user_id', Integer, ForeignKey( "prin_users.principal_id"),
primary_key=True ),
Column('group_id', Integer, ForeignKey( "prin_groups.principal_id"),
primary_key=True ),
)
def testbasic(self):
class Principal(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
class User(Principal):
pass
class Group(Principal):
pass
mapper(Principal, principals)
mapper(User, users, inherits=Principal)
mapper(Group, groups, inherits=Principal, properties={
'users': relationship(User, secondary=user_group_map,
lazy='select', backref="groups")
})
g = Group(name="group1")
g.users.append(User(name="user1", password="pw", email="foo@bar.com", login_id="lg1"))
sess = create_session()
sess.add(g)
sess.flush()
# TODO: put an assertion
class InheritTest2(_base.MappedTest):
"""deals with inheritance and many-to-many relationships"""
@classmethod
def define_tables(cls, metadata):
global foo, bar, foo_bar
foo = Table('foo', metadata,
Column('id', Integer, Sequence('foo_id_seq', optional=True),
primary_key=True),
Column('data', String(20)),
)
bar = Table('bar', metadata,
Column('bid', Integer, ForeignKey('foo.id'), primary_key=True),
#Column('fid', Integer, ForeignKey('foo.id'), )
)
foo_bar = Table('foo_bar', metadata,
Column('foo_id', Integer, ForeignKey('foo.id')),
Column('bar_id', Integer, ForeignKey('bar.bid')))
def testget(self):
class Foo(object):
def __init__(self, data=None):
self.data = data
class Bar(Foo):pass
mapper(Foo, foo)
mapper(Bar, bar, inherits=Foo)
print foo.join(bar).primary_key
print class_mapper(Bar).primary_key
b = Bar('somedata')
sess = create_session()
sess.add(b)
sess.flush()
sess.expunge_all()
# test that "bar.bid" does not need to be referenced in a get
# (ticket 185)
assert sess.query(Bar).get(b.id).id == b.id
def testbasic(self):
class Foo(object):
def __init__(self, data=None):
self.data = data
mapper(Foo, foo)
class Bar(Foo):
pass
mapper(Bar, bar, inherits=Foo, properties={
'foos': relationship(Foo, secondary=foo_bar, lazy='joined')
})
sess = create_session()
b = Bar('barfoo')
sess.add(b)
sess.flush()
f1 = Foo('subfoo1')
f2 = Foo('subfoo2')
b.foos.append(f1)
b.foos.append(f2)
sess.flush()
sess.expunge_all()
l = sess.query(Bar).all()
print l[0]
print l[0].foos
self.assert_unordered_result(l, Bar,
# {'id':1, 'data':'barfoo', 'bid':1, 'foos':(Foo, [{'id':2,'data':'subfoo1'}, {'id':3,'data':'subfoo2'}])},
{'id':b.id, 'data':'barfoo', 'foos':(Foo, [{'id':f1.id,'data':'subfoo1'}, {'id':f2.id,'data':'subfoo2'}])},
)
class InheritTest3(_base.MappedTest):
"""deals with inheritance and many-to-many relationships"""
@classmethod
def define_tables(cls, metadata):
global foo, bar, blub, bar_foo, blub_bar, blub_foo
# the 'data' columns are to appease SQLite which cant handle a blank INSERT
foo = Table('foo', metadata,
Column('id', Integer, Sequence('foo_seq', optional=True),
primary_key=True),
Column('data', String(20)))
bar = Table('bar', metadata,
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
Column('data', String(20)))
blub = Table('blub', metadata,
Column('id', Integer, ForeignKey('bar.id'), primary_key=True),
Column('data', String(20)))
bar_foo = Table('bar_foo', metadata,
Column('bar_id', Integer, ForeignKey('bar.id')),
Column('foo_id', Integer, ForeignKey('foo.id')))
blub_bar = Table('bar_blub', metadata,
Column('blub_id', Integer, ForeignKey('blub.id')),
Column('bar_id', Integer, ForeignKey('bar.id')))
blub_foo = Table('blub_foo', metadata,
Column('blub_id', Integer, ForeignKey('blub.id')),
Column('foo_id', Integer, ForeignKey('foo.id')))
def testbasic(self):
class Foo(object):
def __init__(self, data=None):
self.data = data
def __repr__(self):
return "Foo id %d, data %s" % (self.id, self.data)
mapper(Foo, foo)
class Bar(Foo):
def __repr__(self):
return "Bar id %d, data %s" % (self.id, self.data)
mapper(Bar, bar, inherits=Foo, properties={
'foos' :relationship(Foo, secondary=bar_foo, lazy='select')
})
sess = create_session()
b = Bar('bar #1')
sess.add(b)
b.foos.append(Foo("foo #1"))
b.foos.append(Foo("foo #2"))
sess.flush()
compare = [repr(b)] + sorted([repr(o) for o in b.foos])
sess.expunge_all()
l = sess.query(Bar).all()
print repr(l[0]) + repr(l[0].foos)
found = [repr(l[0])] + sorted([repr(o) for o in l[0].foos])
eq_(found, compare)
@testing.fails_on('maxdb', 'FIXME: unknown')
def testadvanced(self):
class Foo(object):
def __init__(self, data=None):
self.data = data
def __repr__(self):
return "Foo id %d, data %s" % (self.id, self.data)
mapper(Foo, foo)
class Bar(Foo):
def __repr__(self):
return "Bar id %d, data %s" % (self.id, self.data)
mapper(Bar, bar, inherits=Foo)
class Blub(Bar):
def __repr__(self):
return "Blub id %d, data %s, bars %s, foos %s" % (self.id, self.data, repr([b for b in self.bars]), repr([f for f in self.foos]))
mapper(Blub, blub, inherits=Bar, properties={
'bars':relationship(Bar, secondary=blub_bar, lazy='joined'),
'foos':relationship(Foo, secondary=blub_foo, lazy='joined'),
})
sess = create_session()
f1 = Foo("foo #1")
b1 = Bar("bar #1")
b2 = Bar("bar #2")
bl1 = Blub("blub #1")
for o in (f1, b1, b2, bl1):
sess.add(o)
bl1.foos.append(f1)
bl1.bars.append(b2)
sess.flush()
compare = repr(bl1)
blubid = bl1.id
sess.expunge_all()
l = sess.query(Blub).all()
print l
self.assert_(repr(l[0]) == compare)
sess.expunge_all()
x = sess.query(Blub).filter_by(id=blubid).one()
print x
self.assert_(repr(x) == compare)
|