from sqlalchemy.ext import serializer
from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.test import testing
from sqlalchemy import MetaData,Integer,String,ForeignKey,select,desc,func,util
from sqlalchemy.test.schema import Table
from sqlalchemy.test.schema import Column
from sqlalchemy.orm import relationship,sessionmaker,scoped_session,class_mapper,mapper,joinedload,compile_mappers,aliased
from sqlalchemy.test.testing import eq_
from test.orm._base import ComparableEntity,MappedTest
class User(ComparableEntity):
pass
class Address(ComparableEntity):
pass
class SerializeTest(MappedTest):
run_setup_mappers = 'once'
run_inserts = 'once'
run_deletes = None
@classmethod
def define_tables(cls, metadata):
global users, addresses
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50))
)
addresses = Table('addresses', metadata,
Column('id', Integer, primary_key=True),
Column('email', String(50)),
Column('user_id', Integer, ForeignKey('users.id')),
)
@classmethod
def setup_mappers(cls):
global Session
Session = scoped_session(sessionmaker())
mapper(User, users, properties={
'addresses':relationship(Address, backref='user', order_by=addresses.c.id)
})
mapper(Address, addresses)
compile_mappers()
@classmethod
def insert_data(cls):
params = [dict(zip(('id', 'name'), column_values)) for column_values in
[(7, 'jack'),
(8, 'ed'),
(9, 'fred'),
(10, 'chuck')]
]
users.insert().execute(params)
addresses.insert().execute(
[dict(zip(('id', 'user_id', 'email'), column_values)) for column_values in
[(1, 7, "jack@bean.com"),
(2, 8, "ed@wood.com"),
(3, 8, "ed@bettyboop.com"),
(4, 8, "ed@lala.com"),
(5, 9, "fred@fred.com")]
]
)
def test_tables(self):
assert serializer.loads(serializer.dumps(users, -1), users.metadata, Session) is users
def test_columns(self):
assert serializer.loads(serializer.dumps(users.c.name, -1), users.metadata, Session) is users.c.name
def test_mapper(self):
user_mapper = class_mapper(User)
assert serializer.loads(serializer.dumps(user_mapper, -1), None, None) is user_mapper
def test_attribute(self):
assert serializer.loads(serializer.dumps(User.name, -1), None, None) is User.name
def test_expression(self):
expr = select([users]).select_from(users.join(addresses)).limit(5)
re_expr = serializer.loads(serializer.dumps(expr, -1), users.metadata, None)
eq_(
str(expr),
str(re_expr)
)
assert re_expr.bind is testing.db
eq_(
re_expr.execute().fetchall(),
[(7, u'jack'), (8, u'ed'), (8, u'ed'), (8, u'ed'), (9, u'fred')]
)
def test_query(self):
q = Session.query(User).filter(User.name=='ed').options(joinedload(User.addresses))
eq_(q.all(), [User(name='ed', addresses=[Address(id=2), Address(id=3), Address(id=4)])])
q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session)
def go():
eq_(q2.all(), [User(name='ed', addresses=[Address(id=2), Address(id=3), Address(id=4)])])
self.assert_sql_count(testing.db, go, 1)
eq_(q2.join(User.addresses).filter(Address.email=='ed@bettyboop.com').value(func.count('*')), 1)
u1 = Session.query(User).get(8)
q = Session.query(Address).filter(Address.user==u1).order_by(desc(Address.email))
q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session)
eq_(q2.all(), [Address(email='ed@wood.com'), Address(email='ed@lala.com'), Address(email='ed@bettyboop.com')])
q = Session.query(User).join(User.addresses).filter(Address.email.like('%fred%'))
q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session)
eq_(q2.all(), [User(name='fred')])
eq_(list(q2.values(User.id, User.name)), [(9, u'fred')])
@testing.exclude('sqlite', '<=', (3, 5, 9), 'id comparison failing on the buildbot')
def test_aliases(self):
u7, u8, u9, u10 = Session.query(User).order_by(User.id).all()
ualias = aliased(User)
q = Session.query(User, ualias).join((ualias, User.id < ualias.id)).filter(User.id<9).order_by(User.id, ualias.id)
eq_(list(q.all()), [(u7, u8), (u7, u9), (u7, u10), (u8, u9), (u8, u10)])
q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session)
eq_(list(q2.all()), [(u7, u8), (u7, u9), (u7, u10), (u8, u9), (u8, u10)])
def test_any(self):
r = User.addresses.any(Address.email=='x')
ser = serializer.dumps(r, -1)
x = serializer.loads(ser, users.metadata)
eq_(str(r), str(x))
if __name__ == '__main__':
testing.main()
|