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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
import testenv; testenv.configure_for_tests()
import pickle
from testlib import sa, testing
from testlib.sa import Table, Column, Integer, String, ForeignKey
from testlib.sa.orm import mapper, relation, create_session, attributes
from orm import _base, _fixtures
User, EmailUser = None, None
class PickleTest(_fixtures.FixtureTest):
run_inserts = None
@testing.resolve_artifact_names
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.add(u2)
sess.flush()
sess.clear()
self.assertEquals(u1, sess.query(User).get(u2.id))
@testing.resolve_artifact_names
def test_class_deferred_cols(self):
mapper(User, users, properties={
'name':sa.orm.deferred(users.c.name),
'addresses':relation(Address, backref="user")
})
mapper(Address, addresses, properties={
'email_address':sa.orm.deferred(addresses.c.email_address)
})
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='ed@bar.com'))
sess.add(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')]))
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
u2 = sess2.merge(u2, dont_load=True)
self.assertEquals(u2.name, 'ed')
self.assertEquals(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
@testing.resolve_artifact_names
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.add(u1)
sess.flush()
sess.clear()
u1 = sess.query(User).options(sa.orm.defer('name'), sa.orm.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 u2.__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')]))
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
u2 = sess2.merge(u2, dont_load=True)
self.assertEquals(u2.name, 'ed')
assert 'addresses' not in u2.__dict__
ad = u2.addresses[0]
assert 'email_address' in ad.__dict__ # mapper options dont transmit over merge() right now
self.assertEquals(ad.email_address, 'ed@bar.com')
self.assertEquals(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
class PolymorphicDeferredTest(_base.MappedTest):
def define_tables(self, metadata):
Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(30)),
Column('type', String(30)))
Table('email_users', metadata,
Column('id', Integer, ForeignKey('users.id'), primary_key=True),
Column('email_address', String(30)))
def setup_classes(self):
global User, EmailUser
class User(_base.BasicEntity):
pass
class EmailUser(User):
pass
def tearDownAll(self):
global User, EmailUser
User, EmailUser = None, None
_base.MappedTest.tearDownAll(self)
@testing.resolve_artifact_names
def test_polymorphic_deferred(self):
mapper(User, users, polymorphic_identity='user', polymorphic_on=users.c.type)
mapper(EmailUser, email_users, inherits=User, polymorphic_identity='emailuser')
eu = EmailUser(name="user1", email_address='foo@bar.com')
sess = create_session()
sess.add(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')
class CustomSetupTeardowntest(_fixtures.FixtureTest):
@testing.resolve_artifact_names
def test_rebuild_state(self):
"""not much of a 'test', but illustrate how to
remove instance-level state before pickling.
"""
mapper(User, users)
u1 = User()
attributes.manager_of_class(User).teardown_instance(u1)
assert not u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
attributes.manager_of_class(User).setup_instance(u2)
assert attributes.instance_state(u2)
if __name__ == '__main__':
testenv.main()
|
