since these are the only two use cases for the "__props" collection on mapper.
does synonym resolution so also fixes [ticket:598]
for ext in self.extensions:
ext.set(obj, value, previous, initiator or self)
- property = property(lambda s: class_mapper(s.class_).props[s.key],
+ property = property(lambda s: class_mapper(s.class_).get_property(s.key),
doc="the MapperProperty object associated with this attribute")
InstrumentedAttribute.logger = logging.class_logger(InstrumentedAttribute)
except AttributeError:
mapper = context.mapper
for token in self.key.split('.'):
- prop = mapper.props[token]
- if isinstance(prop, SynonymProperty):
- prop = mapper.props[prop.name]
+ prop = mapper.get_property(token, resolve_synonyms=True)
mapper = getattr(prop, 'mapper', None)
self.__prop = prop
return prop
from sqlalchemy.orm import util as mapperutil
from sqlalchemy.orm.util import ExtensionCarrier
from sqlalchemy.orm import sync
-from sqlalchemy.orm.interfaces import MapperProperty, MapperOption, OperationContext, EXT_PASS, MapperExtension
+from sqlalchemy.orm.interfaces import MapperProperty, MapperOption, OperationContext, EXT_PASS, MapperExtension, SynonymProperty
import weakref
__all__ = ['Mapper', 'class_mapper', 'object_mapper', 'mapper_registry']
else:
return False
- def _get_props(self):
+ def get_property(self, key, resolve_synonyms=False, raiseerr=True):
+ """return MapperProperty with the given key."""
self.compile()
- return self.__props
-
- props = property(_get_props, doc="compiles this mapper if needed, and returns the "
- "dictionary of MapperProperty objects associated with this mapper.")
-
+ 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
+
+ def iterate_properties(self):
+ self.compile()
+ return self.__props.itervalues()
+ iterate_properties = property(iterate_properties, doc="returns an iterator of all MapperProperty objects.")
+
def dispose(self):
attribute_manager.reset_class_managed(self.class_)
if hasattr(self.class_, 'c'):
# try to set a LazyLoader on our mapper referencing the parent mapper
mapper = prop.mapper.primary_mapper()
- if not mapper.props.has_key(self.key):
+ if not mapper.get_property(self.key, raiseerr=False) is not None:
pj = self.kwargs.pop('primaryjoin', None)
sj = self.kwargs.pop('secondaryjoin', None)
# the backref property is set on the primary mapper
backref=prop.key, is_backref=True,
**self.kwargs)
mapper._compile_property(self.key, relation);
- elif not isinstance(mapper.props[self.key], PropertyLoader):
+ elif not isinstance(mapper.get_property(self.key), PropertyLoader):
raise exceptions.ArgumentError(
"Can't create backref '%s' on mapper '%s'; an incompatible "
"property of that name already exists" % (self.key, str(mapper)))
else:
# else set one of us as the "backreference"
parent = prop.parent.primary_mapper()
- if parent.class_ is not mapper.props[self.key]._get_target_class():
+ if parent.class_ is not mapper.get_property(self.key)._get_target_class():
raise exceptions.ArgumentError(
"Backrefs do not match: backref '%s' expects to connect to %s, "
"but found a backref already connected to %s" %
- (self.key, str(parent.class_), str(mapper.props[self.key].mapper.class_)))
- if not mapper.props[self.key].is_backref:
+ (self.key, str(parent.class_), str(mapper.get_property(self.key).mapper.class_)))
+ if not mapper.get_property(self.key).is_backref:
prop.is_backref=True
if not prop.viewonly:
prop._dependency_processor.is_backref=True
# reverse_property used by dependencies.ManyToManyDP to check
# association table operations
- prop.reverse_property = mapper.props[self.key]
- mapper.props[self.key].reverse_property = prop
+ prop.reverse_property = mapper.get_property(self.key)
+ mapper.get_property(self.key).reverse_property = prop
def get_extension(self):
"""Return an attribute extension to use with this backreference."""
from sqlalchemy import sql, util, exceptions, sql_util, logging, schema
from sqlalchemy.orm import mapper, class_mapper, object_mapper
-from sqlalchemy.orm.interfaces import OperationContext, SynonymProperty
+from sqlalchemy.orm.interfaces import OperationContext
__all__ = ['Query', 'QueryContext', 'SelectionContext']
"""
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):
joinpoint = self._joinpoint
for key, value in kwargs.iteritems():
- prop = joinpoint.props[key]
+ prop = joinpoint.get_property(key, resolve_synonyms=True)
if isinstance(prop, properties.PropertyLoader):
c = self._with_lazy_criterion(value, prop, True) # & self.join_via(keys[:-1]) - use aliasized join feature
else:
else:
return self.filter(clause)
- def _join_to(self, prop, outerjoin=False, start=None, create_aliases=False):
+ def _join_to(self, keys, outerjoin=False, start=None, create_aliases=False):
if start is None:
start = self._joinpoint
- prop = util.to_list(prop)
- mapper = start
- 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
clause = self._from_obj[-1]
currenttables = [clause]
mapper = start
alias = None
- for key in keys:
- prop = mapper.props[key]
+ for key in util.to_list(keys):
+ 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))
# dont re-join to a table already in our from objects
# 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)
mapper = self._joinpoint
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:
if mapper_ in seen:
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, raiseerr=False)
+ if prop is not None:
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)
merged = self.get(mapper.class_, key[1])
if merged is None:
raise exceptions.AssertionError("Instance %s has an instance key but is not persisted" % mapperutil.instance_str(object))
- for prop in mapper.props.values():
+ for prop in mapper.iterate_properties:
prop.merge(self, object, merged, _recursive)
if key is None:
self.save(merged, entity_name=mapper.entity_name)
def _get_deferred_loader(self, instance, mapper, needs_tables):
def load():
- group = [p for p in mapper.props.values() if isinstance(p.strategy, ColumnLoader) and p.columns[0].table in needs_tables]
+ group = [p for p in mapper.iterate_properties if isinstance(p.strategy, ColumnLoader) and p.columns[0].table in needs_tables]
if self._should_log_debug:
self.logger.debug("deferred load %s group %s" % (mapperutil.attribute_str(instance, self.key), group and ','.join([p.key for p in group]) or 'None'))
if localparent is None:
return None
- prop = localparent.props[self.key]
+ prop = localparent.get_property(self.key)
if prop is not self.parent_property:
return prop._get_strategy(DeferredColumnLoader).setup_loader(instance)
return None
if self.group is not None:
- group = [p for p in localparent.props.values() if isinstance(p.strategy, DeferredColumnLoader) and p.group==self.group]
+ group = [p for p in localparent.iterate_properties if isinstance(p.strategy, DeferredColumnLoader) and p.group==self.group]
else:
group = None
if not mapper.has_mapper(instance):
return None
else:
- prop = mapper.object_mapper(instance).props[self.key]
+ prop = mapper.object_mapper(instance).get_property(self.key)
if prop is not self.parent_property:
return prop._get_strategy(LazyLoader).setup_loader(instance)
def lazyload():
statement.append_from(statement._outerjoin)
- for value in self.select_mapper.props.values():
+ for value in self.select_mapper.iterate_properties:
value.setup(context, eagertable=clauses.eagertarget, parentclauses=clauses, parentmapper=self.select_mapper)
def _create_row_decorator(self, selectcontext, row):
if sess is not None:
if self.cascade is not None and self.cascade.save_update and item not in sess:
mapper = object_mapper(obj)
- prop = mapper.props[self.key]
+ prop = mapper.get_property(self.key)
ename = prop.mapper.entity_name
sess.save_or_update(item, entity_name=ename)
if sess is not None:
if newvalue is not None and self.cascade is not None and self.cascade.save_update and newvalue not in sess:
mapper = object_mapper(obj)
- prop = mapper.props[self.key]
+ prop = mapper.get_property(self.key)
ename = prop.mapper.entity_name
sess.save_or_update(newvalue, entity_name=ename)
)
)
- print str(Person.mapper.props['balls'].primaryjoin)
-
b = Ball('some data')
p = Person('some data')
p.balls.append(b)
)
)
- print str(Person.mapper.props['balls'].primaryjoin)
-
b = Ball('some data')
p = Person('some data')
p.balls.append(b)
mapper(User, users, properties = dict(
addresses = relation(Address, lazy=False, backref=backref('user', lazy=False))
))
- assert class_mapper(User).props['addresses'].lazy is False
- assert class_mapper(Address).props['user'].lazy is False
+ assert class_mapper(User).get_property('addresses').lazy is False
+ assert class_mapper(Address).get_property('user').lazy is False
sess = create_session()
assert fixtures.user_address_result == sess.query(User).all()
m = mapper(User, users, properties = {
'addresses' : relation(mapper(Address, addresses))
}).compile()
- self.assert_(User.addresses.property is m.props['addresses'])
+ self.assert_(User.addresses.property is m.get_property('addresses'))
def testrecursiveselectby(self):
result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_("order 1", "order 2", "order 3")).join(['orders', 'items']).filter_by(id=4).all()
assert [User(id=7, name='jack')] == result
+
+class SynonymTest(QueryTest):
+ keep_mappers = True
+ keep_data = True
+
+ def setup_mappers(self):
+ mapper(User, users, properties={
+ 'name_syn':synonym('name'),
+ 'addresses':relation(Address),
+ 'orders':relation(Order, backref='user'), # o2m, m2o
+ 'orders_syn':synonym('orders')
+ })
+ mapper(Address, addresses)
+ mapper(Order, orders, properties={
+ 'items':relation(Item, secondary=order_items), #m2m
+ 'address':relation(Address), # m2o
+ 'items_syn':synonym('items')
+ })
+ mapper(Item, items, properties={
+ 'keywords':relation(Keyword, secondary=item_keywords) #m2m
+ })
+ mapper(Keyword, keywords)
+
+ def test_joins(self):
+ for j in (
+ ['orders', 'items'],
+ ['orders_syn', 'items'],
+ ['orders', 'items_syn'],
+ ['orders_syn', 'items_syn'],
+ ):
+ result = create_session().query(User).join(j).filter_by(id=3).all()
+ assert [User(id=7, name='jack'), User(id=9, name='fred')] == result
+
+ def test_with_parent(self):
+ for nameprop, orderprop in (
+ ('name', 'orders'),
+ ('name_syn', 'orders'),
+ ('name', 'orders_syn'),
+ ('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).all()
+ assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
+
class InstancesTest(QueryTest):
def test_from_alias(self):
},
)
- assert list(m2.props['sites'].foreign_keys) == [peoplesites.c.person]
+ assert list(m2.get_property('sites').foreign_keys) == [peoplesites.c.person]
p = Person()
p.person = 'im the key'
p.firstname = 'asdf'
collection is being compiled."""
from sqlalchemy import *
+from sqlalchemy.orm import *
import thread, time
from sqlalchemy.orm import mapperlib
from testbase import Table, Column
def get_metadata(self):
return _otest_metadata
def tearDownAll(self):
+ clear_mappers()
_otest_metadata.drop_all()
def tearDown(self):
if not self.keep_mappers:
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
