which resolves synonyms. fixes [ticket:598] for join/join_to/join_via/with_parent
- added synchronization to the mapper() construction step, to avoid
thread collections when pre-existing mappers are compiling in a
different thread [ticket:613]
+ - synonym() properties are fully supported by all Query joining/
+ with_parent operations [ticket:598]
- fixed very stupid bug when deleting items with many-to-many
uselist=False relations
- remember all that stuff about polymorphic_union ? for
from sqlalchemy import sql_util as sqlutil
from sqlalchemy.orm import util as mapperutil
from sqlalchemy.orm import sync
-from sqlalchemy.orm.interfaces import MapperProperty, MapperOption, OperationContext
+from sqlalchemy.orm.interfaces import MapperProperty, MapperOption, OperationContext, SynonymProperty
import weakref
__all__ = ['Mapper', 'MapperExtension', 'class_mapper', 'object_mapper', 'EXT_PASS', 'mapper_registry', 'ExtensionOption']
return self.__props
props = property(_get_props, doc="compiles this mapper if needed, and returns the "
- "dictionary of MapperProperty objects associated with this mapper.")
+ "dictionary of MapperProperty objects associated with this mapper."
+ "(Deprecated; use get_property() and iterate_properties)")
+
+ def get_property(self, key, resolve_synonyms=False, raiseerr=True):
+ """return MapperProperty with the given key.
+
+ forwards compatible with 0.4.
+ """
+
+ self.compile()
+ prop = self.__props.get(key, None)
+ if resolve_synonyms:
+ while isinstance(prop, SynonymProperty):
+ prop = self.__props.get(prop.name, None)
+ if prop is None and raiseerr:
+ raise exceptions.InvalidRequestError("Mapper '%s' has no property '%s'" % (str(self), key))
+ return prop
+
+ iterate_properties = property(lambda self: self._get_props().itervalues(), doc="returns an iterator of all MapperProperty objects."
+ " Forwards compatible with 0.4")
def compile(self):
"""Compile this mapper into its final internal format.
mapper = self.mapper
clause = None
for key in keys:
- prop = mapper.props[key]
+ prop = mapper.get_property(key, resolve_synonyms=True)
if clause is None:
clause = prop.get_join(mapper)
else:
"""
mapper = object_mapper(instance)
- prop = mapper.props[property]
+ prop = mapper.get_property(property, resolve_synonyms=True)
target = prop.mapper
criterion = cls._with_lazy_criterion(instance, prop)
return Query(target, **kwargs).filter(criterion)
from sqlalchemy.orm import properties
mapper = object_mapper(instance)
if property is None:
- for prop in mapper.props.values():
+ for prop in mapper.iterate_properties:
if isinstance(prop, properties.PropertyLoader) and prop.mapper is self.mapper:
break
else:
raise exceptions.InvalidRequestError("Could not locate a property which relates instances of class '%s' to instances of class '%s'" % (self.mapper.class_.__name__, instance.__class__.__name__))
else:
- prop = mapper.props[property]
+ prop = mapper.get_property(property, resolve_synonyms=True)
return self.filter(Query._with_lazy_criterion(instance, prop))
def add_entity(self, entity):
def _join_to(self, prop, outerjoin=False):
if isinstance(prop, list):
- mapper = self._joinpoint
- keys = []
- for key in prop:
- p = mapper.props[key]
- if p._is_self_referential():
- raise exceptions.InvalidRequestError("Self-referential query on '%s' property must be constructed manually using an Alias object for the related table." % (str(p)))
- keys.append(key)
- mapper = p.mapper
+ keys = prop
else:
[keys,p] = self._locate_prop(prop, start=self._joinpoint)
clause = self._from_obj[-1]
mapper = self._joinpoint
for key in keys:
- prop = mapper.props[key]
+ prop = mapper.get_property(key, resolve_synonyms=True)
if prop._is_self_referential():
raise exceptions.InvalidRequestError("Self-referential query on '%s' property must be constructed manually using an Alias object for the related table." % str(prop))
if outerjoin:
return None
seen.add(mapper_)
if mapper_.props.has_key(key):
- prop = mapper_.props[key]
- if isinstance(prop, SynonymProperty):
- prop = mapper_.props[prop.name]
+ prop = mapper_.get_property(key, resolve_synonyms=True)
if isinstance(prop, properties.PropertyLoader):
keys.insert(0, prop.key)
return prop
else:
- for prop in mapper_.props.values():
+ for prop in mapper_.iterate_properties:
if not isinstance(prop, properties.PropertyLoader):
continue
x = search_for_prop(prop.mapper)
# give all the attached properties a chance to modify the query
# TODO: doing this off the select_mapper. if its the polymorphic mapper, then
# it has no relations() on it. should we compile those too into the query ? (i.e. eagerloads)
- for value in self.select_mapper.props.values():
+ for value in self.select_mapper.iterate_properties:
value.setup(context)
# additional entities/columns, add those to selection criterion
if isinstance(m, type):
m = mapper.class_mapper(m)
if isinstance(m, mapper.Mapper):
- for value in m.props.values():
+ for value in m.iterate_properties:
value.setup(context)
elif isinstance(m, sql.ColumnElement):
statement.append_column(m)
"""test the with_parent()) method and one-to-many relationships"""
m = mapper(User, users, properties={
+ 'user_name_syn':synonym('user_name'),
'orders':relation(mapper(Order, orders, properties={
- 'items':relation(mapper(Item, orderitems))
- }))
+ 'items':relation(mapper(Item, orderitems)),
+ 'items_syn':synonym('items')
+ })),
+ 'orders_syn':synonym('orders')
})
sess = create_session()
assert False
except exceptions.InvalidRequestError, e:
assert str(e) == "Could not locate a property which relates instances of class 'Item' to instances of class 'User'"
+
+
+ for nameprop, orderprop in (
+ ('user_name', 'orders'),
+ ('user_name_syn', 'orders'),
+ ('user_name', 'orders_syn'),
+ ('user_name_syn', 'orders_syn'),
+ ):
+ sess = create_session()
+ q = sess.query(User)
+
+ u1 = q.filter_by(**{nameprop:'jack'}).one()
+
+ o = sess.query(Order).with_parent(u1, property=orderprop).list()
+ self.assert_result(o, Order, *user_all_result[0]['orders'][1])
def testwithparentm2m(self):
"""test the with_parent() method and many-to-many relationships"""
self.assert_result(k, Keyword, *item_keyword_result[1]['keywords'][1])
- def testautojoin(self):
+ def test_join(self):
"""test functions derived from Query's _join_to function."""
m = mapper(User, users, properties={
'orders':relation(mapper(Order, orders, properties={
- 'items':relation(mapper(Item, orderitems))
- }))
+ 'items':relation(mapper(Item, orderitems)),
+ 'items_syn':synonym('items')
+ })),
+
+ 'orders_syn':synonym('orders'),
})
sess = create_session()
q = sess.query(m)
- l = q.filter(orderitems.c.item_name=='item 4').join(['orders', 'items']).list()
- self.assert_result(l, User, user_result[0])
-
+ for j in (
+ ['orders', 'items'],
+ ['orders', 'items_syn'],
+ ['orders_syn', 'items'],
+ ['orders_syn', 'items_syn'],
+ ):
+ for q in (
+ q.filter(orderitems.c.item_name=='item 4').join(j),
+ q.filter(orderitems.c.item_name=='item 4').join(j[-1]),
+ q.filter(orderitems.c.item_name=='item 4').filter(q.join_via(j)),
+ q.filter(orderitems.c.item_name=='item 4').filter(q.join_to(j[-1])),
+ ):
+ l = q.all()
+ self.assert_result(l, User, user_result[0])
+
l = q.select_by(item_name='item 4')
self.assert_result(l, User, user_result[0])
l = q.select(q.join_to('orders'), order_by=desc(orders.c.user_id), limit=2, offset=1)
self.assert_result(l, User, *(user_all_result[2], user_all_result[0]))
- l = q.select(q.join_to('addresses'), order_by=desc(addresses.c.email_address), limit=1, offset=0)
- self.assert_result(l, User, *(user_all_result[0],))
-
def testonetoone(self):
m = mapper(User, users, properties = dict(
address = relation(mapper(Address, addresses), lazy = False, uselist = False)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
