mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-06-01 05:18:44 -04:00
Mike Bayer
285 lines
7.8 KiB
Python
285 lines
7.8 KiB
Python
from sqlalchemy import Integer
|
|
from sqlalchemy.types import PickleType, TypeDecorator, VARCHAR
|
|
from sqlalchemy.orm import mapper, Session, composite
|
|
from sqlalchemy.orm.mapper import Mapper
|
|
from sqlalchemy.orm.instrumentation import ClassManager
|
|
from test.lib.schema import Table, Column
|
|
from test.lib.testing import eq_
|
|
from test.lib.util import picklers
|
|
from test.lib import testing
|
|
from test.orm import _base
|
|
import sys
|
|
import pickle
|
|
|
|
class Foo(_base.BasicEntity):
|
|
pass
|
|
|
|
class _MutableDictTestBase(object):
|
|
run_define_tables = 'each'
|
|
|
|
@classmethod
|
|
def _type_fixture(cls):
|
|
from sqlalchemy.ext.mutable import Mutable
|
|
|
|
# needed for pickle support
|
|
global MutationDict
|
|
|
|
class MutationDict(Mutable, dict):
|
|
@classmethod
|
|
def coerce(cls, key, value):
|
|
if not isinstance(value, MutationDict):
|
|
if isinstance(value, dict):
|
|
return MutationDict(value)
|
|
return Mutable.coerce(key, value)
|
|
else:
|
|
return value
|
|
|
|
def __getstate__(self):
|
|
return dict(self)
|
|
|
|
def __setstate__(self, state):
|
|
self.update(state)
|
|
|
|
def __setitem__(self, key, value):
|
|
dict.__setitem__(self, key, value)
|
|
self.changed()
|
|
|
|
def __delitem__(self, key):
|
|
dict.__delitem__(self, key)
|
|
self.changed()
|
|
return MutationDict
|
|
|
|
@testing.resolve_artifact_names
|
|
def setup_mappers(cls):
|
|
mapper(Foo, foo)
|
|
|
|
def teardown(self):
|
|
# clear out mapper events
|
|
Mapper.dispatch._clear()
|
|
ClassManager.dispatch._clear()
|
|
super(_MutableDictTestBase, self).teardown()
|
|
|
|
@testing.resolve_artifact_names
|
|
def test_in_place_mutation(self):
|
|
sess = Session()
|
|
|
|
f1 = Foo(data={'a':'b'})
|
|
sess.add(f1)
|
|
sess.commit()
|
|
|
|
f1.data['a'] = 'c'
|
|
sess.commit()
|
|
|
|
eq_(f1.data, {'a':'c'})
|
|
|
|
@testing.resolve_artifact_names
|
|
def test_pickle_parent(self):
|
|
sess = Session()
|
|
|
|
f1 = Foo(data={'a':'b'})
|
|
sess.add(f1)
|
|
sess.commit()
|
|
f1.data
|
|
sess.close()
|
|
|
|
for loads, dumps in picklers():
|
|
sess = Session()
|
|
f2 = loads(dumps(f1))
|
|
sess.add(f2)
|
|
f2.data['a'] = 'c'
|
|
assert f2 in sess.dirty
|
|
|
|
@testing.resolve_artifact_names
|
|
def _test_non_mutable(self):
|
|
sess = Session()
|
|
|
|
f1 = Foo(non_mutable_data={'a':'b'})
|
|
sess.add(f1)
|
|
sess.commit()
|
|
|
|
f1.non_mutable_data['a'] = 'c'
|
|
sess.commit()
|
|
|
|
eq_(f1.non_mutable_data, {'a':'b'})
|
|
|
|
class MutableWithScalarPickleTest(_MutableDictTestBase, _base.MappedTest):
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
MutationDict = cls._type_fixture()
|
|
|
|
Table('foo', metadata,
|
|
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
|
Column('data', MutationDict.as_mutable(PickleType)),
|
|
Column('non_mutable_data', PickleType)
|
|
)
|
|
|
|
def test_non_mutable(self):
|
|
self._test_non_mutable()
|
|
|
|
class MutableWithScalarJSONTest(_MutableDictTestBase, _base.MappedTest):
|
|
# json introduced in 2.6
|
|
__skip_if__ = lambda : sys.version_info < (2, 6),
|
|
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
import json
|
|
|
|
class JSONEncodedDict(TypeDecorator):
|
|
impl = VARCHAR(50)
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is not None:
|
|
value = json.dumps(value)
|
|
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is not None:
|
|
value = json.loads(value)
|
|
return value
|
|
|
|
MutationDict = cls._type_fixture()
|
|
|
|
Table('foo', metadata,
|
|
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
|
Column('data', MutationDict.as_mutable(JSONEncodedDict)),
|
|
Column('non_mutable_data', JSONEncodedDict)
|
|
)
|
|
|
|
def test_non_mutable(self):
|
|
self._test_non_mutable()
|
|
|
|
class MutableAssociationScalarPickleTest(_MutableDictTestBase, _base.MappedTest):
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
MutationDict = cls._type_fixture()
|
|
MutationDict.associate_with(PickleType)
|
|
|
|
Table('foo', metadata,
|
|
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
|
Column('data', PickleType)
|
|
)
|
|
|
|
class MutableAssociationScalarJSONTest(_MutableDictTestBase, _base.MappedTest):
|
|
# json introduced in 2.6
|
|
__skip_if__ = lambda : sys.version_info < (2, 6),
|
|
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
import json
|
|
|
|
class JSONEncodedDict(TypeDecorator):
|
|
impl = VARCHAR(50)
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is not None:
|
|
value = json.dumps(value)
|
|
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is not None:
|
|
value = json.loads(value)
|
|
return value
|
|
|
|
MutationDict = cls._type_fixture()
|
|
MutationDict.associate_with(JSONEncodedDict)
|
|
|
|
Table('foo', metadata,
|
|
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
|
Column('data', JSONEncodedDict)
|
|
)
|
|
|
|
|
|
class _CompositeTestBase(object):
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
Table('foo', metadata,
|
|
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
|
Column('x', Integer),
|
|
Column('y', Integer)
|
|
)
|
|
|
|
def teardown(self):
|
|
# clear out mapper events
|
|
Mapper.dispatch._clear()
|
|
ClassManager.dispatch._clear()
|
|
super(_CompositeTestBase, self).teardown()
|
|
|
|
class MutableCompositesTest(_CompositeTestBase, _base.MappedTest):
|
|
|
|
@classmethod
|
|
def _type_fixture(cls):
|
|
|
|
from sqlalchemy.ext.mutable import Mutable
|
|
from sqlalchemy.ext.mutable import MutableComposite
|
|
|
|
global Point
|
|
|
|
class Point(MutableComposite):
|
|
def __init__(self, x, y):
|
|
self.x = x
|
|
self.y = y
|
|
|
|
def __setattr__(self, key, value):
|
|
object.__setattr__(self, key, value)
|
|
self.changed()
|
|
|
|
def __composite_values__(self):
|
|
return self.x, self.y
|
|
|
|
def __getstate__(self):
|
|
d = dict(self.__dict__)
|
|
d.pop('_parents', None)
|
|
return d
|
|
|
|
#def __setstate__(self, state):
|
|
# self.x, self.y = state
|
|
|
|
def __eq__(self, other):
|
|
return isinstance(other, Point) and \
|
|
other.x == self.x and \
|
|
other.y == self.y
|
|
return Point
|
|
|
|
@classmethod
|
|
@testing.resolve_artifact_names
|
|
def setup_mappers(cls):
|
|
Point = cls._type_fixture()
|
|
|
|
mapper(Foo, foo, properties={
|
|
'data':composite(Point, foo.c.x, foo.c.y)
|
|
})
|
|
|
|
@testing.resolve_artifact_names
|
|
def test_in_place_mutation(self):
|
|
sess = Session()
|
|
d = Point(3, 4)
|
|
f1 = Foo(data=d)
|
|
sess.add(f1)
|
|
sess.commit()
|
|
|
|
f1.data.y = 5
|
|
sess.commit()
|
|
|
|
eq_(f1.data, Point(3, 5))
|
|
|
|
@testing.resolve_artifact_names
|
|
def test_pickle_of_parent(self):
|
|
sess = Session()
|
|
d = Point(3, 4)
|
|
f1 = Foo(data=d)
|
|
sess.add(f1)
|
|
sess.commit()
|
|
|
|
f1.data
|
|
assert 'data' in f1.__dict__
|
|
sess.close()
|
|
|
|
for loads, dumps in picklers():
|
|
sess = Session()
|
|
f2 = loads(dumps(f1))
|
|
sess.add(f2)
|
|
f2.data.y = 12
|
|
assert f2 in sess.dirty
|