return True
else:
return False
-
+
+ def _distance(self, othercolumn):
+ c = othercolumn
+ count = 0
+ while c is not self:
+ c = c._source_column
+ if c is None:
+ return -1
+ count += 1
+ return count
+
def _make_proxy(self, selectable, name=None):
"""Create a new ``ColumnElement`` representing this
``ColumnElement`` as it appears in the select list of a
"""
if column in self.c:
return column
-
+
if require_embedded and column not in util.Set(self._get_all_embedded_columns()):
if not raiseerr:
return None
for co in self._adjusted_exportable_columns():
cp = self._proxy_column(co)
for ci in cp.orig_set:
- # note that some ambiguity is raised here, whereby a selectable might have more than
- # one column that maps to an "original" column. examples include unions and joins
- self._orig_cols[ci] = cp
+ cx = self._orig_cols.get(ci)
+ if cx is None or ci._distance(cp) < ci._distance(cx):
+ self._orig_cols[ci] = cp
if self.oid_column is not None:
for ci in self.oid_column.orig_set:
self._orig_cols[ci] = self.oid_column
self.type = sqltypes.to_instance(totype)
self.clause = clause
self.typeclause = _TypeClause(self.type)
-
+ self._source_column = None
+
def get_children(self, **kwargs):
return self.clause, self.typeclause
def accept_visitor(self, visitor):
def _make_proxy(self, selectable, name=None):
if name is not None:
co = _ColumnClause(name, selectable, type=self.type)
+ co._source_column = self
co.orig_set = self.orig_set
selectable.columns[name]= co
return co
self.table = selectable
self.type = sqltypes.to_instance(type)
self._is_oid = _is_oid
+ self._source_column = None
self.__label = None
self.case_sensitive = case_sensitive
self.is_literal = is_literal
is_literal = self.is_literal and (name is None or name == self.name)
c = _ColumnClause(name or self.name, selectable=selectable, _is_oid=self._is_oid, type=self.type, is_literal=is_literal)
c.orig_set = self.orig_set
+ c._source_column = self
if not self._is_oid:
selectable.columns[c.name] = c
return c
# object is not in the session; therefore the lazy load cant trigger here,
# eager load had to succeed
assert len([c for c in d2.comments]) == 1
+
+class EagerTest6(testbase.ORMTest):
+ def define_tables(self, metadata):
+ global project_t, task_t, task_status_t, task_type_t, message_t, message_type_t
-
+ project_t = Table('prj', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('created', DateTime , ),
+ Column('title', Unicode(100)),
+ )
+
+ task_t = Table('task', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('status_id', Integer, ForeignKey('task_status.id'), nullable=False),
+ Column('title', Unicode(100)),
+ Column('task_type_id', Integer , ForeignKey('task_type.id'), nullable=False),
+ Column('prj_id', Integer , ForeignKey('prj.id'), nullable=False),
+ )
+
+ task_status_t = Table('task_status', metadata,
+ Column('id', Integer, primary_key=True),
+ )
+
+ task_type_t = Table('task_type', metadata,
+ Column('id', Integer, primary_key=True),
+ )
+
+ message_t = Table('msg', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('posted', DateTime, index=True,),
+ Column('type_id', Integer, ForeignKey('msg_type.id')),
+ Column('task_id', Integer, ForeignKey('task.id')),
+ )
+
+ message_type_t = Table('msg_type', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('name', Unicode(20)),
+ Column('display_name', Unicode(20)),
+ )
+
+ def setUp(self):
+ testbase.db.execute("INSERT INTO prj (title) values('project 1');")
+ testbase.db.execute("INSERT INTO task_status (id) values(1);")
+ testbase.db.execute("INSERT INTO task_type(id) values(1);")
+ testbase.db.execute("INSERT INTO task (title, task_type_id, status_id, prj_id) values('task 1',1,1,1);")
+
+ def test_nested_joins(self):
+ # this is testing some subtle column resolution stuff,
+ # concerning corresponding_column() being extremely accurate
+ # as well as how mapper sets up its column properties
+
+ class Task(object):pass
+ class Task_Type(object):pass
+ class Message(object):pass
+ class Message_Type(object):pass
+
+ tsk_cnt_join = outerjoin(project_t, task_t, task_t.c.prj_id==project_t.c.id)
+
+ ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')],
+ from_obj=[tsk_cnt_join], group_by=[project_t.c.id]).alias('prj_tsk_cnt_s')
+ j = join(project_t, ss, project_t.c.id == ss.c.prj_id)
+
+ mapper(Task_Type, task_type_t)
+
+ mapper( Task, task_t,
+ properties=dict(type=relation(Task_Type, lazy=False),
+ ))
+
+ mapper(Message_Type, message_type_t)
+
+ mapper(Message, message_t,
+ properties=dict(type=relation(Message_Type, lazy=False, uselist=False),
+ ))
+
+ tsk_cnt_join = outerjoin(project_t, task_t, task_t.c.prj_id==project_t.c.id)
+ ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')],
+ from_obj=[tsk_cnt_join], group_by=[project_t.c.id]).alias('prj_tsk_cnt_s')
+ j = join(project_t, ss, project_t.c.id == ss.c.prj_id)
+
+ j = outerjoin( task_t, message_t, task_t.c.id==message_t.c.task_id)
+ jj = select([ task_t.c.id.label('task_id'),
+ func.count(message_t.c.id).label('props_cnt')],
+ from_obj=[j], group_by=[task_t.c.id]).alias('prop_c_s')
+ jjj = join(task_t, jj, task_t.c.id == jj.c.task_id)
+
+ class cls(object):pass
+
+ props =dict(type=relation(Task_Type, lazy=False))
+ print [c.key for c in jjj.c]
+ cls.mapper = mapper( cls, jjj, properties=props)
+
+ session = create_session()
+
+ for t in session.query(cls.mapper).limit(10).offset(0).list():
+ print t.id, t.title, t.props_cnt
+
if __name__ == "__main__":
testbase.main()
)\r
\r
class SelectableTest(testbase.AssertMixin):\r
+ def testdistance(self):\r
+ s = select([table.c.col1.label('c2'), table.c.col1, table.c.col1.label('c1')])\r
+\r
+ # didnt do this yet...col.label().make_proxy() has same "distance" as col.make_proxy() so far\r
+ #assert s.corresponding_column(table.c.col1) is s.c.col1\r
+ assert s.corresponding_column(s.c.col1) is s.c.col1\r
+ assert s.corresponding_column(s.c.c1) is s.c.c1\r
+ \r
def testjoinagainstself(self):\r
jj = select([table.c.col1.label('bar_col1')])\r
jjj = join(table, jj, table.c.col1==jj.c.bar_col1)\r
+ \r
+ # test column directly agaisnt itself\r
assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1\r
\r
+ assert jjj.corresponding_column(jj.c.bar_col1) is jjj.c.bar_col1\r
+ \r
+ # test alias of the join, targets the column with the least \r
+ # "distance" between the requested column and the returned column\r
+ # (i.e. there is less indirection between j2.c.table1_col1 and table.c.col1, than\r
+ # there is from j2.c.bar_col1 to table.c.col1)\r
+ j2 = jjj.alias('foo')\r
+ assert j2.corresponding_column(table.c.col1) is j2.c.table1_col1\r
+ \r
+\r
def testjoinagainstjoin(self):\r
j = outerjoin(table, table2, table.c.col1==table2.c.col2)\r
jj = select([ table.c.col1.label('bar_col1')],from_obj=[j]).alias('foo')\r
jjj = join(table, jj, table.c.col1==jj.c.bar_col1)\r
assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1\r
+\r
+ j2 = jjj.alias('foo')\r
+ print j2.corresponding_column(jjj.c.table1_col1)\r
+ assert j2.corresponding_column(jjj.c.table1_col1) is j2.c.table1_col1\r
\r
+ assert jjj.corresponding_column(jj.c.bar_col1) is jj.c.bar_col1\r
\r
def testtablealias(self):\r
a = table.alias('a')\r
j = join(a, table2)\r
\r
criterion = a.c.col1 == table2.c.col2\r
- print\r
- print str(j)\r
+ print criterion\r
+ print j.onclause\r
self.assert_(criterion.compare(j.onclause))\r
\r
def testselectlabels(self):\r
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
