or remove operation is received on the
now-detached collection. [ticket:2476]
+ - [bug] Declarative can now propagate a column
+ declared on a single-table inheritance subclass
+ up to the parent class' table, when the parent
+ class is itself mapped to a join() or select()
+ statement, directly or via joined inheritane,
+ and not just a Table. [ticket:2549]
+
- [bug] An error is emitted when uselist=False
is combined with a "dynamic" loader.
This is a warning in 0.7.9.
elif inherits:
inherited_mapper = _declared_mapping_info(inherits)
inherited_table = inherited_mapper.local_table
+ inherited_mapped_table = inherited_mapper.mapped_table
if table is None:
# single table inheritance.
(c, cls, inherited_table.c[c.name])
)
inherited_table.append_column(c)
+ if inherited_mapped_table is not None and \
+ inherited_mapped_table is not inherited_table:
+ inherited_mapped_table._refresh_for_new_column(c)
mt = _MapperConfig(mapper_cls,
cls, table,
class _MapperConfig(object):
configs = util.OrderedDict()
+ mapped_table = None
def __init__(self, mapper_cls,
cls,
return o[0](self, expr, op, other, reverse=True, *o[1:], **kwargs)
-
def _check_literal(self, expr, operator, other):
if isinstance(other, BindParameter) and \
isinstance(other.type, sqltypes.NullType):
self.primary_key = ColumnSet()
self.foreign_keys = set()
+ @property
+ def _cols_populated(self):
+ return '_columns' in self.__dict__
+
def _populate_column_collection(self):
- pass
+ """Called on subclasses to establish the .c collection.
+
+ Each implementation has a different way of establishing
+ this collection.
+
+ """
+
+ def _refresh_for_new_column(self, column):
+ """Given a column added to the .c collection of an underlying
+ selectable, produce the local version of that column, assuming this
+ selectable ultimately should proxy this column.
+
+ this is used to "ping" a derived selectable to add a new column
+ to its .c. collection when a Column has been added to one of the
+ Table objects it ultimtely derives from.
+
+ If the given selectable hasn't populated it's .c. collection yet,
+ it should at least pass on the message to the contained selectables,
+ but it will return None.
+
+ This method is currently used by Declarative to allow Table
+ columns to be added to a partially constructed inheritance
+ mapping that may have already produced joins. The method
+ isn't public right now, as the full span of implications
+ and/or caveats aren't yet clear.
+
+ It's also possible that this functionality could be invoked by
+ default via an event, which would require that
+ selectables maintain a weak referencing collection of all
+ derivations.
+
+ """
+ if not self._cols_populated:
+ return None
+ elif column.key in self.columns and self.columns[column.key] is column:
+ return column
+ else:
+ return None
class BindParameter(ColumnElement):
"""Represent a bind parameter.
self.foreign_keys.update(itertools.chain(
*[col.foreign_keys for col in columns]))
+ def _refresh_for_new_column(self, column):
+ col = self.left._refresh_for_new_column(column)
+ if col is None:
+ col = self.right._refresh_for_new_column(column)
+ if col is not None:
+ if self._cols_populated:
+ self._columns[col._label] = col
+ self.foreign_keys.add(col)
+ if col.primary_key:
+ self.primary_key.add(col)
+ return col
+ return None
+
def _copy_internals(self, clone=_clone, **kw):
self._reset_exported()
self.left = clone(self.left, **kw)
for col in self.element.columns:
col._make_proxy(self)
+ def _refresh_for_new_column(self, column):
+ col = self.element._refresh_for_new_column(column)
+ if col is not None:
+ if not self._cols_populated:
+ return None
+ else:
+ return col._make_proxy(self)
+ else:
+ return None
+
def _copy_internals(self, clone=_clone, **kw):
# don't apply anything to an aliased Table
# for now. May want to drive this from
proxy.proxies = [c._annotate({'weight': i + 1}) for (i,
c) in enumerate(cols)]
+ def _refresh_for_new_column(self, column):
+ for s in self.selects:
+ s._refresh_for_new_column(column)
+
+ if not self._cols_populated:
+ return None
+
+ raise NotImplementedError("CompoundSelect constructs don't support "
+ "addition of columns to underlying selectables")
+
def _copy_internals(self, clone=_clone, **kw):
self._reset_exported()
self.selects = [clone(s, **kw) for s in self.selects]
name=c._label if self.use_labels else None,
key=c._key_label if self.use_labels else None)
+ def _refresh_for_new_column(self, column):
+ for fromclause in self._froms:
+ col = fromclause._refresh_for_new_column(column)
+ if col is not None:
+ if col in self.inner_columns and self._cols_populated:
+ our_label = col._key_label if self.use_labels else col.key
+ if our_label not in self.c:
+ return col._make_proxy(self,
+ name=col._label if self.use_labels else None,
+ key=col._key_label if self.use_labels else None)
+ return None
+ return None
+
def self_group(self, against=None):
"""return a 'grouping' construct as per the ClauseElement
specification.
eq_(sess.query(Engineer).filter_by(primary_language='cobol'
).one(), Engineer(name='vlad', primary_language='cobol'))
+ def test_single_from_joined_colsonsub(self):
+ class Person(Base, fixtures.ComparableEntity):
+
+ __tablename__ = 'people'
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ name = Column(String(50))
+ discriminator = Column('type', String(50))
+ __mapper_args__ = {'polymorphic_on': discriminator}
+
+ class Manager(Person):
+ __tablename__ = 'manager'
+ __mapper_args__ = {'polymorphic_identity': 'manager'}
+ id = Column(Integer, ForeignKey('people.id'), primary_key=True)
+ golf_swing = Column(String(50))
+
+ class Boss(Manager):
+ boss_name = Column(String(50))
+
+ is_(
+ Boss.__mapper__.column_attrs['boss_name'].columns[0],
+ Manager.__table__.c.boss_name
+ )
+
def test_polymorphic_on_converted_from_inst(self):
class A(Base):
__tablename__ = 'A'
Table('t1', MetaData(), c1)
eq_(c1._label, "t1_c1")
+
+class RefreshForNewColTest(fixtures.TestBase):
+ def test_join_uninit(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ j = a.join(b, a.c.x == b.c.y)
+
+ q = column('q')
+ b.append_column(q)
+ j._refresh_for_new_column(q)
+ assert j.c.b_q is q
+
+ def test_join_init(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ j = a.join(b, a.c.x == b.c.y)
+ j.c
+ q = column('q')
+ b.append_column(q)
+ j._refresh_for_new_column(q)
+ assert j.c.b_q is q
+
+
+ def test_join_samename_init(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ j = a.join(b, a.c.x == b.c.y)
+ j.c
+ q = column('x')
+ b.append_column(q)
+ j._refresh_for_new_column(q)
+ assert j.c.b_x is q
+
+ def test_select_samename_init(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ s = select([a, b]).apply_labels()
+ s.c
+ q = column('x')
+ b.append_column(q)
+ s._refresh_for_new_column(q)
+ assert q in s.c.b_x.proxy_set
+
+ def test_aliased_select_samename_uninit(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ s = select([a, b]).apply_labels().alias()
+ q = column('x')
+ b.append_column(q)
+ s._refresh_for_new_column(q)
+ assert q in s.c.b_x.proxy_set
+
+ def test_aliased_select_samename_init(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ s = select([a, b]).apply_labels().alias()
+ s.c
+ q = column('x')
+ b.append_column(q)
+ s._refresh_for_new_column(q)
+ assert q in s.c.b_x.proxy_set
+
+ def test_aliased_select_irrelevant(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ c = table('c', column('z'))
+ s = select([a, b]).apply_labels().alias()
+ s.c
+ q = column('x')
+ c.append_column(q)
+ s._refresh_for_new_column(q)
+ assert 'c_x' not in s.c
+
+ def test_aliased_select_no_cols_clause(self):
+ a = table('a', column('x'))
+ s = select([a.c.x]).apply_labels().alias()
+ s.c
+ q = column('q')
+ a.append_column(q)
+ s._refresh_for_new_column(q)
+ assert 'a_q' not in s.c
+
+ def test_union_uninit(self):
+ a = table('a', column('x'))
+ s1 = select([a])
+ s2 = select([a])
+ s3 = s1.union(s2)
+ q = column('q')
+ a.append_column(q)
+ s3._refresh_for_new_column(q)
+ assert a.c.q in s3.c.q.proxy_set
+
+ def test_union_init_raises(self):
+ a = table('a', column('x'))
+ s1 = select([a])
+ s2 = select([a])
+ s3 = s1.union(s2)
+ s3.c
+ q = column('q')
+ a.append_column(q)
+ assert_raises_message(
+ NotImplementedError,
+ "CompoundSelect constructs don't support addition of "
+ "columns to underlying selectables",
+ s3._refresh_for_new_column, q
+ )
+ def test_nested_join_uninit(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ c = table('c', column('z'))
+ j = a.join(b, a.c.x == b.c.y).join(c, b.c.y == c.c.z)
+
+ q = column('q')
+ b.append_column(q)
+ j._refresh_for_new_column(q)
+ assert j.c.b_q is q
+
+ def test_nested_join_init(self):
+ a = table('a', column('x'))
+ b = table('b', column('y'))
+ c = table('c', column('z'))
+ j = a.join(b, a.c.x == b.c.y).join(c, b.c.y == c.c.z)
+
+ j.c
+ q = column('q')
+ b.append_column(q)
+ j._refresh_for_new_column(q)
+ assert j.c.b_q is q
+
class AnonLabelTest(fixtures.TestBase):
"""Test behaviors fixed by [ticket:2168]."""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
