of being deferred until later. This mimics the old 0.3
behavior.
+ - fixed bug in polymorphic inheritance which made it
+ difficult to set a working "order_by" on a polymorphic
+ mapper
+
+ - fixed a rather expensive call in Query that was slowing
+ down polymorphic queries
+
- general
- warnings are now issued as type exceptions.SAWarning.
if self.version_id_col is None:
self.version_id_col = self.inherits.version_id_col
-
+
+ for mapper in self.iterate_to_root():
+ if hasattr(mapper, '_genned_equivalent_columns'):
+ del mapper._genned_equivalent_columns
+
if self.order_by is False:
self.order_by = self.inherits.order_by
self.polymorphic_map = self.inherits.polymorphic_map
if self.inherits is not None and not self.concrete and not self.primary_key_argument:
self.primary_key = self.inherits.primary_key
self._get_clause = self.inherits._get_clause
- self._equivalent_columns = {}
else:
# create the "primary_key" for this mapper. this will flatten "equivalent" primary key columns
# into one column, where "equivalent" means that one column references the other via foreign key, or
# multiple columns that all reference a common parent column. it will also resolve the column
# against the "mapped_table" of this mapper.
- self._equivalent_columns = self._get_equivalent_columns()
primary_key = expression.ColumnSet()
_get_clause.clauses.append(primary_key == bind)
self._get_clause = (_get_clause, _get_params)
- def _get_equivalent_columns(self):
+ def __get_equivalent_columns(self):
"""Create a map of all *equivalent* columns, based on
the determination of column pairs that are equated to
one another either by an established foreign key relationship
equivs(col, util.Set(), col)
return result
-
+ def _equivalent_columns(self):
+ if hasattr(self, '_genned_equivalent_columns'):
+ return self._genned_equivalent_columns
+ else:
+ self._genned_equivalent_columns = self.__get_equivalent_columns()
+ return self._genned_equivalent_columns
+ _equivalent_columns = property(_equivalent_columns)
+
class _CompileOnAttr(PropComparator):
"""placeholder class attribute which fires mapper compilation on access"""
"""
if self.select_table is not self.mapped_table:
- props = {}
+ self.__surrogate_mapper = Mapper(self.class_, self.select_table, non_primary=True, _polymorphic_map=self.polymorphic_map, polymorphic_on=_corresponding_column_or_error(self.select_table, self.polymorphic_on), primary_key=self.primary_key_argument)
+ adapter = sqlutil.ClauseAdapter(self.select_table, equivalents=self.__surrogate_mapper._equivalent_columns)
+
+ if self.order_by:
+ order_by = [expression._literal_as_text(o) for o in util.to_list(self.order_by) or []]
+ order_by = adapter.copy_and_process(order_by)
+ self.__surrogate_mapper.order_by=order_by
+
if self._init_properties is not None:
for key, prop in self._init_properties.iteritems():
if expression.is_column(prop):
- props[key] = _corresponding_column_or_error(self.select_table, prop)
+ self.__surrogate_mapper.add_property(key, _corresponding_column_or_error(self.select_table, prop))
elif (isinstance(prop, list) and expression.is_column(prop[0])):
- props[key] = [_corresponding_column_or_error(self.select_table, c) for c in prop]
- self.__surrogate_mapper = Mapper(self.class_, self.select_table, non_primary=True, properties=props, _polymorphic_map=self.polymorphic_map, polymorphic_on=_corresponding_column_or_error(self.select_table, self.polymorphic_on), primary_key=self.primary_key_argument)
+ self.__surrogate_mapper.add_property(key, [_corresponding_column_or_error(self.select_table, c) for c in prop])
+
def _compile_class(self):
"""If this mapper is to be a primary mapper (i.e. the
# as we will be using the polymorphic selectables (i.e. select_table argument to Mapper) to figure this out,
# first create maps of all the "equivalent" columns, since polymorphic selectables will often munge
# several "equivalent" columns (such as parent/child fk cols) into just one column.
- target_equivalents = self.mapper._get_equivalent_columns()
+ target_equivalents = self.mapper._equivalent_columns
if self.secondaryjoin:
self.polymorphic_secondaryjoin = ClauseAdapter(self.mapper.select_table).traverse(self.secondaryjoin, clone=True)
try:
return self._parent_join_cache[(parent, primary, secondary, polymorphic_parent)]
except KeyError:
- parent_equivalents = parent._get_equivalent_columns()
+ parent_equivalents = parent._equivalent_columns
secondaryjoin = self.polymorphic_secondaryjoin
if polymorphic_parent:
# adapt the "parent" side of our join condition to the "polymorphic" select of the parent
# adapt for poylmorphic mapper
# TODO: generalize the polymorphic mapper adaption to that of the select_from() adaption
if not adapt_criterion and whereclause is not None and (self.mapper is not self.select_mapper):
- whereclause = sql_util.ClauseAdapter(from_obj, equivalents=self.select_mapper._get_equivalent_columns()).traverse(whereclause)
+ whereclause = sql_util.ClauseAdapter(from_obj, equivalents=self.select_mapper._equivalent_columns).traverse(whereclause)
# TODO: mappers added via add_entity(), adapt their queries also,
# if those mappers are polymorphic
order_by = self._order_by
if order_by is False:
- order_by = self.mapper.order_by
+ order_by = self.select_mapper.order_by
if order_by is False:
order_by = []
if self.table.default_order_by() is not None:
context.primary_columns.append(c)
statement = sql.select(context.primary_columns + context.secondary_columns, whereclause, from_obj=from_obj, use_labels=True, for_update=for_update, order_by=util.to_list(order_by), **self._select_args())
-
+
if context.eager_joins:
if adapt_criterion:
context.eager_joins = sql_util.ClauseAdapter(from_obj).traverse(context.eager_joins)
mapper(Company, companies, properties={
'employees':relation(Person)
})
- mapper(Person, people, select_table=person_join, polymorphic_on=people.c.type, polymorphic_identity='person', order_by=person_join.c.person_id,
+
+ # testing a order_by here as well; the surrogate mapper has to adapt it
+ mapper(Person, people, select_table=person_join, polymorphic_on=people.c.type, polymorphic_identity='person', order_by=people.c.person_id,
properties={
'paperwork':relation(Paperwork)
})
mapper(Paperwork, paperwork)
def insert_data(self):
+ global all_employees, c1_employees, c2_employees, e1, e2, b1, m1, e3
+
c1 = Company(name="MegaCorp, Inc.")
c2 = Company(name="Elbonia, Inc.")
e1 = Engineer(name="dilbert", engineer_name="dilbert", primary_language="java", status="regular engineer", paperwork=[
sess.flush()
sess.clear()
- global all_employees, c1_employees, c2_employees
all_employees = [e1, e2, b1, m1, e3]
c1_employees = [e1, e2, b1, m1]
c2_employees = [e3]
+ def test_filter_on_subclass(self):
+ print Manager.person_id == Engineer.person_id
+ print Manager.c.person_id == Engineer.c.person_id
+
+ sess = create_session()
+ self.assertEquals(sess.query(Engineer).all()[0], Engineer(name="dilbert"))
+
+ self.assertEquals(sess.query(Engineer).first(), Engineer(name="dilbert"))
+
+ self.assertEquals(sess.query(Engineer).filter(Engineer.person_id==e1.person_id).first(), Engineer(name="dilbert"))
+
+ self.assertEquals(sess.query(Manager).filter(Manager.person_id==m1.person_id).one(), Manager(name="dogbert"))
+
+ self.assertEquals(sess.query(Manager).filter(Manager.person_id==b1.person_id).one(), Boss(name="pointy haired boss"))
+
+ self.assertEquals(sess.query(Boss).filter(Boss.person_id==b1.person_id).one(), Boss(name="pointy haired boss"))
+
def test_load_all(self):
sess = create_session()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
