1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
import testbase
from sqlalchemy import *
from sqlalchemy import exceptions
from sqlalchemy.orm import *
from testlib import *
from testlib.fixtures import *
import pickle
class EmailUser(User):
pass
class PickleTest(FixtureTest):
keep_mappers = False
keep_data = False
def test_transient(self):
mapper(User, users, properties={
'addresses':relation(Address, backref="user")
})
mapper(Address, addresses)
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='ed@bar.com'))
u2 = pickle.loads(pickle.dumps(u1))
sess.save(u2)
sess.flush()
sess.clear()
self.assertEquals(u1, sess.query(User).get(u2.id))
def test_class_deferred_cols(self):
mapper(User, users, properties={
'name':deferred(users.c.name),
'addresses':relation(Address, backref="user")
})
mapper(Address, addresses, properties={
'email_address':deferred(addresses.c.email_address)
})
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='ed@bar.com'))
sess.save(u1)
sess.flush()
sess.clear()
u1 = sess.query(User).get(u1.id)
assert 'name' not in u1.__dict__
assert 'addresses' not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
sess2.update(u2)
self.assertEquals(u2.name, 'ed')
self.assertEquals(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
def test_instance_deferred_cols(self):
mapper(User, users, properties={
'addresses':relation(Address, backref="user")
})
mapper(Address, addresses)
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='ed@bar.com'))
sess.save(u1)
sess.flush()
sess.clear()
u1 = sess.query(User).options(defer('name'), defer('addresses.email_address')).get(u1.id)
assert 'name' not in u1.__dict__
assert 'addresses' not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
sess2.update(u2)
self.assertEquals(u2.name, 'ed')
assert 'addresses' not in u1.__dict__
ad = u2.addresses[0]
assert 'email_address' not in ad.__dict__
self.assertEquals(ad.email_address, 'ed@bar.com')
self.assertEquals(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
class PolymorphicDeferredTest(ORMTest):
def define_tables(self, metadata):
global users, email_users
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(30)),
Column('type', String(30)),
)
email_users = Table('email_users', metadata,
Column('id', Integer, ForeignKey('users.id'), primary_key=True),
Column('email_address', String(30))
)
def test_polymorphic_deferred(self):
mapper(User, users, polymorphic_identity='user', polymorphic_on=users.c.type, polymorphic_fetch='deferred')
mapper(EmailUser, email_users, inherits=User, polymorphic_identity='emailuser')
eu = EmailUser(name="user1", email_address='foo@bar.com')
sess = create_session()
sess.save(eu)
sess.flush()
sess.clear()
eu = sess.query(User).first()
eu2 = pickle.loads(pickle.dumps(eu))
sess2 = create_session()
sess2.update(eu2)
assert 'email_address' not in eu2.__dict__
self.assertEquals(eu2.email_address, 'foo@bar.com')
if __name__ == '__main__':
testbase.main()
|
