parent, owning_class, target_class, value_attr
)
- is_object = getattr(target_class, value_attr).impl.uses_objects
+ attr = getattr(target_class, value_attr)
+ if attr._is_internal_proxy and not hasattr(attr, "impl"):
+ return AmbiguousAssociationProxyInstance(
+ parent, owning_class, target_class, value_attr
+ )
+ is_object = attr.impl.uses_objects
if is_object:
return ObjectAssociationProxyInstance(
parent, owning_class, target_class, value_attr
)
def get(self, obj):
- self._ambiguous()
-
- def set(self, obj, values):
- self._ambiguous()
-
- def delete(self, obj):
- self._ambiguous()
+ if obj is None:
+ self._ambiguous()
+ else:
+ return super(AmbiguousAssociationProxyInstance, self).get(obj)
def any(self, criterion=None, **kwargs):
self._ambiguous()
if parent_instance is not None:
actual_obj = getattr(parent_instance, self.target_collection)
if actual_obj is not None:
- instance_class = type(actual_obj)
- if instance_class not in self._lookup_cache:
- self._populate_cache(instance_class)
-
try:
- return self._lookup_cache[instance_class]
- except KeyError:
+ insp = inspect(actual_obj)
+ except exc.NoInspectionAvailable:
pass
+ else:
+ mapper = insp.mapper
+ instance_class = mapper.class_
+ if instance_class not in self._lookup_cache:
+ self._populate_cache(instance_class, mapper)
+
+ try:
+ return self._lookup_cache[instance_class]
+ except KeyError:
+ pass
# no object or ambiguous object given, so return "self", which
- # is a no-op proxy.
+ # is a proxy with generally only instance-level functionality
return self
- def _populate_cache(self, instance_class):
+ def _populate_cache(self, instance_class, mapper):
prop = orm.class_mapper(self.owning_class).get_property(
self.target_collection
)
- if inspect(instance_class).mapper.isa(prop.mapper):
+ if mapper.isa(prop.mapper):
target_class = instance_class
try:
target_assoc = self._cls_unwrap_target_assoc_proxy(
**kw
)
+ def _assert_raises_attribute(self, message, fn, *arg, **kw):
+ assert_raises_message(AttributeError, message, fn, *arg, **kw)
+
def test_column_collection_expressions(self):
B, C, C2 = self.classes("B", "C", "C2")
D, C, C2 = self.classes("D", "C", "C2")
d1 = D()
- self._assert_raises_ambiguous(getattr, d1, "c_data")
+ eq_(d1.c_data, None)
def test_subclass_only_owner_assign(self):
D, C, C2 = self.classes("D", "C", "C2")
D, C, C2 = self.classes("D", "C", "C2")
d1 = D()
- self._assert_raises_ambiguous(getattr, d1, "c_data")
+ eq_(d1.c_data, None)
def test_subclass_only_owner_delete(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C2(csub_only_data="some c2"))
+ eq_(d1.c.csub_only_data, "some c2")
del d1.c_data
+ assert not hasattr(d1.c, "csub_only_data")
- self._assert_raises_ambiguous(getattr, d1, "c_data")
-
- def test_subclass_only_owner_assign_raises(self):
+ def test_subclass_only_owner_assign_passes(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C())
- self._assert_raises_ambiguous(setattr, d1, "c_data", "some c1")
+ d1.c_data = "some c1"
+
+ # not mapped, but we set it
+ eq_(d1.c.csub_only_data, "some c1")
def test_subclass_only_owner_get_raises(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C())
- self._assert_raises_ambiguous(getattr, d1, "c_data")
+ self._assert_raises_attribute(
+ "'C' object has no attribute 'csub_only_data'",
+ getattr,
+ d1,
+ "c_data",
+ )
def test_subclass_only_owner_delete_raises(self):
D, C, C2 = self.classes("D", "C", "C2")
# now switch
d1.c = C()
- self._assert_raises_ambiguous(delattr, d1, "c_data")
+ self._assert_raises_attribute("csub_only_data", delattr, d1, "c_data")
def test_subclasses_conflicting_types(self):
B, D, C, C1, C2 = self.classes("B", "D", "C", "C1", "C2")
self._assert_raises_ambiguous(lambda: D.c_data.any(B.id == 5))
+class ProxyAttrTest(fixtures.DeclarativeMappedTest):
+ @classmethod
+ def setup_classes(cls):
+ from sqlalchemy.ext.hybrid import hybrid_property
+ from sqlalchemy.orm.interfaces import PropComparator
+
+ Base = cls.DeclarativeBasic
+
+ class A(Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ bs = relationship("B")
+
+ b_data = association_proxy("bs", "value")
+ well_behaved_b_data = association_proxy("bs", "well_behaved_value")
+
+ class B(Base):
+ __tablename__ = "b"
+
+ id = Column(Integer, primary_key=True)
+ aid = Column(ForeignKey("a.id"))
+ data = Column(String(50))
+
+ @hybrid_property
+ def well_behaved_value(self):
+ return self.data
+
+ @well_behaved_value.setter
+ def well_behaved_value(self, value):
+ self.data = value
+
+ @hybrid_property
+ def value(self):
+ return self.data
+
+ @value.setter
+ def value(self, value):
+ self.data = value
+
+ @value.comparator
+ class value(PropComparator):
+ # comparator has no proxy __getattr__, so we can't
+ # get to impl to see what we ar proxying towards.
+ def __init__(self, cls):
+ self.cls = cls
+
+ def test_get_ambiguous(self):
+ A, B = self.classes("A", "B")
+
+ a1 = A(bs=[B(data="b1")])
+ eq_(a1.b_data[0], "b1")
+
+ def test_get_nonambiguous(self):
+ A, B = self.classes("A", "B")
+
+ a1 = A(bs=[B(data="b1")])
+ eq_(a1.well_behaved_b_data[0], "b1")
+
+ def test_set_ambiguous(self):
+ A, B = self.classes("A", "B")
+
+ a1 = A(bs=[B()])
+
+ a1.b_data[0] = "b1"
+ eq_(a1.b_data[0], "b1")
+
+ def test_set_nonambiguous(self):
+ A, B = self.classes("A", "B")
+
+ a1 = A(bs=[B()])
+
+ a1.b_data[0] = "b1"
+ eq_(a1.well_behaved_b_data[0], "b1")
+
+ def test_expr_nonambiguous(self):
+ A, B = self.classes("A", "B")
+
+ eq_(
+ str(A.well_behaved_b_data == 5),
+ "EXISTS (SELECT 1 \nFROM a, b \nWHERE "
+ "a.id = b.aid AND b.data = :data_1)",
+ )
+
+ def test_expr_ambiguous(self):
+ A, B = self.classes("A", "B")
+
+ assert_raises_message(
+ AttributeError,
+ "Association proxy A.bs refers to an attribute "
+ "'value' that is not directly mapped",
+ getattr,
+ A,
+ "b_data",
+ )
+
+
class ScopeBehaviorTest(fixtures.DeclarativeMappedTest):
# test some GC scenarios, including issue #4268
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
