*(tuple(cols) + tuple(args)), **table_kw)
else:
table = cls.__table__
-
+ if cols:
+ raise exceptions.ArgumentError("Can't add additional columns when specifying __table__")
+
mapper_args = getattr(cls, '__mapper_args__', {})
if 'inherits' not in mapper_args:
inherits = cls.__mro__[1]
mapper_args['exclude_properties'] = exclude_properties = \
set([c.key for c in inherited_table.c if c not in inherited_mapper._columntoproperty])
exclude_properties.difference_update([c.key for c in cols])
-
+
cls.__mapper__ = mapper_cls(cls, table, properties=our_stuff, **mapper_args)
class DeclarativeMeta(type):
if col in pks:
if history.deleted:
params[col._label] = prop.get_col_value(col, history.deleted[0])
+ hasdata = True
else:
# row switch logic can reach us here
# remove the pk from the update params so the update doesn't
# attempt to include the pk in the update statement
del params[col.key]
params[col._label] = prop.get_col_value(col, history.added[0])
- hasdata = True
+ else:
+ hasdata = True
elif col in pks:
params[col._label] = mapper._get_state_attr_by_column(state, col)
if hasdata:
"""Describes an object attribute that corresponds to a table column."""
def __init__(self, *columns, **kwargs):
- """The list of `columns` describes a single object
- property. If there are multiple tables joined together for the
- mapper, this list represents the equivalent column as it
- appears across each table.
- """
+ """Construct a ColumnProperty.
+
+ :param \*columns: The list of `columns` describes a single
+ object property. If there are multiple tables joined
+ together for the mapper, this list represents the equivalent
+ column as it appears across each table.
+
+ :param group:
+
+ :param deferred:
+
+ :param comparator_factory:
+ :param descriptor:
+
+ :param extension:
+
+ """
self.columns = [expression._labeled(c) for c in columns]
self.group = kwargs.pop('group', None)
self.deferred = kwargs.pop('deferred', False)
self.comparator_factory = kwargs.pop('comparator_factory', self.__class__.Comparator)
self.descriptor = kwargs.pop('descriptor', None)
self.extension = kwargs.pop('extension', None)
+ if kwargs:
+ raise TypeError(
+ "%s received unexpected keyword argument(s): %s" % (
+ self.__class__.__name__, ', '.join(sorted(kwargs.keys()))))
+
util.set_creation_order(self)
if self.no_instrument:
self.strategy_class = strategies.UninstrumentedColumnLoader
mapper.SynonymProperty = SynonymProperty
mapper.ComparableProperty = ComparableProperty
mapper.RelationProperty = RelationProperty
-mapper.ConcreteInheritedProperty = ConcreteInheritedProperty
\ No newline at end of file
+mapper.ConcreteInheritedProperty = ConcreteInheritedProperty
"Unknown arguments passed to Column: " + repr(kwargs.keys()))
def __str__(self):
- if self.table is not None:
+ if self.name is None:
+ return "(no name)"
+ elif self.table is not None:
if self.table.named_with_column:
return (self.table.description + "." + self.description)
else:
else:
return self.description
+ @property
def bind(self):
return self.table.bind
- bind = property(bind)
def references(self, column):
"""Return True if this Column references the given column via foreign key."""
class User(Base):
id = Column('id', Integer, primary_key=True)
self.assertRaisesMessage(sa.exc.InvalidRequestError, "does not have a __table__", go)
+
+ def test_cant_add_columns(self):
+ t = Table('t', Base.metadata, Column('id', Integer, primary_key=True))
+ def go():
+ class User(Base):
+ __table__ = t
+ foo = Column(Integer, primary_key=True)
+ self.assertRaisesMessage(sa.exc.ArgumentError, "add additional columns", go)
+
+ def test_undefer_column_name(self):
+ # TODO: not sure if there was an explicit
+ # test for this elsewhere
+ foo = Column(Integer)
+ eq_(str(foo), '(no name)')
+ eq_(foo.key, None)
+ eq_(foo.name, None)
+ decl._undefer_column_name('foo', foo)
+ eq_(str(foo), 'foo')
+ eq_(foo.key, 'foo')
+ eq_(foo.name, 'foo')
def test_recompile_on_othermapper(self):
"""declarative version of the same test in mappers.py"""
assert list(sess.execute(t5.select(), mapper=T5)) == [(2, 'some other t5')]
assert list(sess.execute(t6.select(), mapper=T5)) == [(1, 'some other t6', 2)]
+class InheritingRowSwitchTest(_base.MappedTest):
+ def define_tables(self, metadata):
+ Table('parent', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('pdata', String(30))
+ )
+ Table('child', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('pid', Integer, ForeignKey('parent.id')),
+ Column('cdata', String(30))
+ )
+
+ def setup_classes(self):
+ class P(_base.ComparableEntity):
+ pass
+
+ class C(P):
+ pass
+
+ @testing.resolve_artifact_names
+ def test_row_switch_no_child_table(self):
+ mapper(P, parent)
+ mapper(C, child, inherits=P)
+
+ sess = create_session()
+ c1 = C(id=1, pdata='c1', cdata='c1')
+ sess.add(c1)
+ sess.flush()
+
+ # establish a row switch between c1 and c2.
+ # c2 has no value for the "child" table
+ c2 = C(id=1, pdata='c2')
+ sess.add(c2)
+ sess.delete(c1)
+
+ self.assert_sql_execution(testing.db, sess.flush,
+ CompiledSQL("UPDATE parent SET pdata=:pdata WHERE parent.id = :parent_id",
+ {'pdata':'c2', 'parent_id':1}
+ )
+ )
+
+
+
class TransactionTest(_base.MappedTest):
__requires__ = ('deferrable_constraints',)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
