1
from sqlalchemy.test.testing import eq_, assert_raises, assert_raises_message
1
from test.lib.testing import eq_, assert_raises, assert_raises_message
3
3
from sqlalchemy import util
4
import sqlalchemy.orm.attributes as attributes
4
from sqlalchemy.orm import attributes, instrumentation
5
5
from sqlalchemy.orm.collections import collection
6
from sqlalchemy.orm.attributes import set_attribute, get_attribute, del_attribute, is_instrumented
6
from sqlalchemy.orm.attributes import set_attribute, get_attribute, del_attribute
7
from sqlalchemy.orm.instrumentation import is_instrumented
7
8
from sqlalchemy.orm import clear_mappers
8
9
from sqlalchemy.orm import InstrumentationManager
9
from sqlalchemy.test import *
10
from test.orm import _base
10
from test.lib import *
11
from test.lib import fixtures
12
13
class MyTypesManager(InstrumentationManager):
63
64
__sa_instrumentation_manager__ = staticmethod(__sa_instrumentation_manager__)
65
66
# This proves SA can handle a class with non-string dict keys
67
if not util.pypy and not util.jython:
67
68
locals()[42] = 99 # Don't remove this line!
71
70
def __init__(self, **kwargs):
100
99
del self._goofy_dict[key]
102
class UserDefinedExtensionTest(_base.ORMTest):
101
class UserDefinedExtensionTest(fixtures.ORMTest):
104
103
def teardown_class(cls):
106
attributes._install_lookup_strategy(util.symbol('native'))
105
instrumentation._install_lookup_strategy(util.symbol('native'))
108
107
def test_instance_dict(self):
109
108
class User(MyClass):
112
attributes.register_class(User)
111
instrumentation.register_class(User)
113
112
attributes.register_attribute(User, 'user_id', uselist = False, useobject=False)
114
113
attributes.register_attribute(User, 'user_name', uselist = False, useobject=False)
115
114
attributes.register_attribute(User, 'email_address', uselist = False, useobject=False)
125
124
class User(base):
128
attributes.register_class(User)
127
instrumentation.register_class(User)
129
128
attributes.register_attribute(User, 'user_id', uselist = False, useobject=False)
130
129
attributes.register_attribute(User, 'user_name', uselist = False, useobject=False)
131
130
attributes.register_attribute(User, 'email_address', uselist = False, useobject=False)
153
152
state.dict[k] = data[k]
154
153
return attributes.ATTR_WAS_SET
156
attributes.register_class(Foo)
157
manager = attributes.manager_of_class(Foo)
155
manager = instrumentation.register_class(Foo)
158
156
manager.deferred_scalar_loader = loader
159
157
attributes.register_attribute(Foo, 'a', uselist=False, useobject=False)
160
158
attributes.register_attribute(Foo, 'b', uselist=False, useobject=False)
162
assert Foo in attributes.instrumentation_registry._state_finders
160
assert Foo in instrumentation.instrumentation_registry._state_finders
164
attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
162
attributes.instance_state(f).expire(attributes.instance_dict(f), set())
165
163
eq_(f.a, "this is a")
168
166
f.a = "this is some new a"
169
attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
167
attributes.instance_state(f).expire(attributes.instance_dict(f), set())
170
168
eq_(f.a, "this is a")
173
attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
171
attributes.instance_state(f).expire(attributes.instance_dict(f), set())
174
172
f.a = "this is another new a"
175
173
eq_(f.a, "this is another new a")
178
attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
176
attributes.instance_state(f).expire(attributes.instance_dict(f), set())
179
177
eq_(f.a, "this is a")
194
192
class Foo(base):pass
195
193
class Bar(Foo):pass
197
attributes.register_class(Foo)
198
attributes.register_class(Bar)
195
instrumentation.register_class(Foo)
196
instrumentation.register_class(Bar)
198
def func1(state, passive):
202
199
return "this is the foo attr"
200
def func2(state, passive):
205
201
return "this is the bar attr"
202
def func3(state, passive):
208
203
return "this is the shared attr"
209
attributes.register_attribute(Foo, 'element', uselist=False, callable_=lambda o:func1, useobject=True)
210
attributes.register_attribute(Foo, 'element2', uselist=False, callable_=lambda o:func3, useobject=True)
211
attributes.register_attribute(Bar, 'element', uselist=False, callable_=lambda o:func2, useobject=True)
204
attributes.register_attribute(Foo, 'element',
205
uselist=False, callable_=func1,
207
attributes.register_attribute(Foo, 'element2',
208
uselist=False, callable_=func3,
210
attributes.register_attribute(Bar, 'element',
211
uselist=False, callable_=func2,
222
223
class Post(base):pass
223
224
class Blog(base):pass
225
attributes.register_class(Post)
226
attributes.register_class(Blog)
227
attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True)
228
attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), trackparent=True, useobject=True)
226
instrumentation.register_class(Post)
227
instrumentation.register_class(Blog)
228
attributes.register_attribute(Post, 'blog', uselist=False,
229
backref='posts', trackparent=True, useobject=True)
230
attributes.register_attribute(Blog, 'posts', uselist=True,
231
backref='blog', trackparent=True, useobject=True)
230
233
(p1, p2, p3) = (Post(), Post(), Post())
231
234
b.posts.append(p1)
259
attributes.register_class(Foo)
260
attributes.register_class(Bar)
262
instrumentation.register_class(Foo)
263
instrumentation.register_class(Bar)
261
264
attributes.register_attribute(Foo, "name", uselist=False, useobject=False)
262
265
attributes.register_attribute(Foo, "bars", uselist=True, trackparent=True, useobject=True)
263
266
attributes.register_attribute(Bar, "name", uselist=False, useobject=False)
291
294
def test_null_instrumentation(self):
292
295
class Foo(MyBaseClass):
294
attributes.register_class(Foo)
297
instrumentation.register_class(Foo)
295
298
attributes.register_attribute(Foo, "name", uselist=False, useobject=False)
296
299
attributes.register_attribute(Foo, "bars", uselist=True, trackparent=True, useobject=True)
304
307
class Unknown(object): pass
305
308
class Known(MyBaseClass): pass
307
attributes.register_class(Known)
310
instrumentation.register_class(Known)
308
311
k, u = Known(), Unknown()
310
assert attributes.manager_of_class(Unknown) is None
311
assert attributes.manager_of_class(Known) is not None
312
assert attributes.manager_of_class(None) is None
313
assert instrumentation.manager_of_class(Unknown) is None
314
assert instrumentation.manager_of_class(Known) is not None
315
assert instrumentation.manager_of_class(None) is None
314
317
assert attributes.instance_state(k) is not None
315
318
assert_raises((AttributeError, KeyError),