directives in statements. Courtesy
Diana Clarke [ticket:2443]
+ - [bug] Fixed bug in 0.7.6 introduced by
+ [ticket:2409] whereby column_mapped_collection
+ used against columns that were mapped as
+ joins or other indirect selectables
+ would fail to function.
+
- [feature] Added new flag to @validates
include_removes. When True, collection
remove and attribute del events
from sqlalchemy.sql import expression
from sqlalchemy import schema, util, exc as sa_exc
-
-
__all__ = ['collection', 'collection_adapter',
'mapped_collection', 'column_mapped_collection',
'attribute_mapped_collection']
__instrumentation_mutex = util.threading.Lock()
+
+class _PlainColumnGetter(object):
+ """Plain column getter, stores collection of Column objects
+ directly.
+
+ Serializes to a :class:`._SerializableColumnGetterV2`
+ which has more expensive __call__() performance
+ and some rare caveats.
+
+ """
+ def __init__(self, cols):
+ self.cols = cols
+ self.composite = len(cols) > 1
+
+ def __reduce__(self):
+ return _SerializableColumnGetterV2._reduce_from_cols(self.cols)
+
+ def _cols(self, mapper):
+ return self.cols
+
+ def __call__(self, value):
+ state = instance_state(value)
+ m = _state_mapper(state)
+
+ key = [
+ m._get_state_attr_by_column(state, state.dict, col)
+ for col in self._cols(m)
+ ]
+
+ if self.composite:
+ return tuple(key)
+ else:
+ return key[0]
+
class _SerializableColumnGetter(object):
+ """Column-based getter used in version 0.7.6 only.
+
+ Remains here for pickle compatibility with 0.7.6.
+
+ """
def __init__(self, colkeys):
self.colkeys = colkeys
self.composite = len(colkeys) > 1
-
def __reduce__(self):
return _SerializableColumnGetter, (self.colkeys,)
-
def __call__(self, value):
state = instance_state(value)
m = _state_mapper(state)
else:
return key[0]
+class _SerializableColumnGetterV2(_PlainColumnGetter):
+ """Updated serializable getter which deals with
+ multi-table mapped classes.
+
+ Two extremely unusual cases are not supported.
+ Mappings which have tables across multiple metadata
+ objects, or which are mapped to non-Table selectables
+ linked across inheriting mappers may fail to function
+ here.
+
+ """
+
+ def __init__(self, colkeys):
+ self.colkeys = colkeys
+ self.composite = len(colkeys) > 1
+
+ def __reduce__(self):
+ return self.__class__, (self.colkeys,)
+
+ @classmethod
+ def _reduce_from_cols(cls, cols):
+ def _table_key(c):
+ if not isinstance(c.table, expression.TableClause):
+ return None
+ else:
+ return c.table.key
+ colkeys = [(c.key, _table_key(c)) for c in cols]
+ return _SerializableColumnGetterV2, (colkeys,)
+
+ def _cols(self, mapper):
+ cols = []
+ metadata = getattr(mapper.local_table, 'metadata', None)
+ for (ckey, tkey) in self.colkeys:
+ if tkey is None or \
+ metadata is None or \
+ tkey not in metadata:
+ cols.append(mapper.local_table.c[ckey])
+ else:
+ cols.append(metadata.tables[tkey].c[ckey])
+ return cols
+
+
def column_mapped_collection(mapping_spec):
"""A dictionary-based collection type with column-based keying.
from sqlalchemy.orm.util import _state_mapper
from sqlalchemy.orm.attributes import instance_state
- cols = [c.key for c in [
- expression._only_column_elements(q, "mapping_spec")
- for q in util.to_list(mapping_spec)]]
- keyfunc = _SerializableColumnGetter(cols)
+ cols = [expression._only_column_elements(q, "mapping_spec")
+ for q in util.to_list(mapping_spec)
+ ]
+ keyfunc = _PlainColumnGetter(cols)
return lambda: MappedCollection(keyfunc)
class _SerializableAttrGetter(object):
class Foo(BaseObject):
__tablename__ = "foo"
- id = Column(Integer(), primary_key=True, test_needs_autoincrement=True)
+ id = Column(Integer(), primary_key=True)
bar_id = Column(Integer, ForeignKey('bar.id'))
- class Bar(BaseObject):
- __tablename__ = "bar"
- id = Column(Integer(), primary_key=True, test_needs_autoincrement=True)
- foos = relationship(Foo, collection_class=collections.column_mapped_collection(Foo.id))
- foos2 = relationship(Foo, collection_class=collections.column_mapped_collection((Foo.id, Foo.bar_id)))
-
- eq_(Bar.foos.property.collection_class().keyfunc(Foo(id=3)), 3)
- eq_(Bar.foos2.property.collection_class().keyfunc(Foo(id=3, bar_id=12)), (3, 12))
+ for spec, obj, expected in (
+ (Foo.id, Foo(id=3), 3),
+ ((Foo.id, Foo.bar_id), Foo(id=3, bar_id=12), (3, 12))
+ ):
+ eq_(
+ collections.column_mapped_collection(spec)().keyfunc(obj),
+ expected
+ )
def test_column_mapped_assertions(self):
assert_raises_message(sa_exc.ArgumentError,
collection_class = lambda: Ordered2(lambda v: (v.a, v.b))
self._test_composite_mapped(collection_class)
+class ColumnMappedWSerialize(fixtures.MappedTest):
+ """test the column_mapped_collection serializer against
+ multi-table and indirect table edge cases, including
+ serialization."""
+
+ run_create_tables = run_deletes = None
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('foo', metadata,
+ Column('id', Integer(), primary_key=True),
+ Column('b', String(128))
+ )
+ Table('bar', metadata,
+ Column('id', Integer(), primary_key=True),
+ Column('foo_id', Integer, ForeignKey('foo.id')),
+ Column('bat_id', Integer),
+ schema="x"
+ )
+ @classmethod
+ def setup_classes(cls):
+ class Foo(cls.Basic):
+ pass
+ class Bar(Foo):
+ pass
+
+ def test_indirect_table_column_mapped(self):
+ Foo = self.classes.Foo
+ Bar = self.classes.Bar
+ bar = self.tables["x.bar"]
+ mapper(Foo, self.tables.foo, properties={
+ "foo_id":self.tables.foo.c.id
+ })
+ mapper(Bar, bar, inherits=Foo, properties={
+ "bar_id":bar.c.id,
+ })
+
+ bar_spec = Bar(foo_id=1, bar_id=2, bat_id=3)
+ self._run_test([
+ (Foo.foo_id, bar_spec, 1),
+ ((Bar.bar_id, Bar.bat_id), bar_spec, (2, 3)),
+ (Bar.foo_id, bar_spec, 1),
+ (bar.c.id, bar_spec, 2),
+ ])
+
+ def test_selectable_column_mapped(self):
+ from sqlalchemy import select
+ s = select([self.tables.foo]).alias()
+ Foo = self.classes.Foo
+ mapper(Foo, s)
+ self._run_test([
+ (Foo.b, Foo(b=5), 5),
+ (s.c.b, Foo(b=5), 5)
+ ])
+
+ def _run_test(self, specs):
+ from test.lib.util import picklers
+ for spec, obj, expected in specs:
+ coll = collections.column_mapped_collection(spec)()
+ eq_(
+ coll.keyfunc(obj),
+ expected
+ )
+ # ensure we do the right thing with __reduce__
+ for loads, dumps in picklers():
+ c2 = loads(dumps(coll))
+ eq_(c2.keyfunc(obj), expected)
+ c3 = loads(dumps(c2))
+ eq_(c3.keyfunc(obj), expected)
+
class CustomCollectionsTest(fixtures.MappedTest):
"""test the integration of collections with mapped classes."""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
