.. changelog::
:version: 0.9.5
+ .. change::
+ :tags: bug, orm
+ :tickets: 3042
+ :versions: 1.0.0
+
+ Additional checks have been added for the case where an inheriting
+ mapper is implicitly combining one of its column-based attributes
+ with that of the parent, where those columns normally don't necessarily
+ share the same value. This is an extension of an existing check that
+ was added via :ticket:`1892`; however this new check emits only a
+ warning, instead of an exception, to allow for applications that may
+ be relying upon the existing behavior.
+
+ .. seealso::
+
+ :ref:`faq_combining_columns`
+
.. change::
:tags: bug, sql
:tickets: 3023
this differs from :attr:`.Mapper.mapped_table` in the case of a mapper mapped
using inheritance to a composed selectable.
+.. _faq_combining_columns:
+
+I'm getting a warning or error about "Implicitly combining column X under attribute Y"
+--------------------------------------------------------------------------------------
+
+This condition refers to when a mapping contains two columns that are being
+mapped under the same attribute name due to their name, but there's no indication
+that this is intentional. A mapped class needs to have explicit names for
+every attribute that is to store an independent value; when two columns have the
+same name and aren't disambiguated, they fall under the same attribute and
+the effect is that the value from one column is **copied** into the other, based
+on which column was assigned to the attribute first.
+
+This behavior is often desirable and is allowed without warning in the case
+where the two columns are linked together via a foreign key relationship
+within an inheritance mapping. When the warning or exception occurs, the
+issue can be resolved by either assigning the columns to differently-named
+attributes, or if combining them together is desired, by using
+:func:`.column_property` to make this explicit.
+
+Given the example as follows::
+
+ from sqlalchemy import Integer, Column, ForeignKey
+ from sqlalchemy.ext.declarative import declarative_base
+
+ Base = declarative_base()
+
+ class A(Base):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+
+ class B(A):
+ __tablename__ = 'b'
+
+ id = Column(Integer, primary_key=True)
+ a_id = Column(Integer, ForeignKey('a.id'))
+
+As of SQLAlchemy version 0.9.5, the above condition is detected, and will
+warn that the ``id`` column of ``A`` and ``B`` is being combined under
+the same-named attribute ``id``, which above is a serious issue since it means
+that a ``B`` object's primary key will always mirror that of its ``A``.
+
+A mapping which resolves this is as follows::
+
+ class A(Base):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+
+ class B(A):
+ __tablename__ = 'b'
+
+ b_id = Column('id', Integer, primary_key=True)
+ a_id = Column(Integer, ForeignKey('a.id'))
+
+Suppose we did want ``A.id`` and ``B.id`` to be mirrors of each other, despite
+the fact that ``B.a_id`` is where ``A.id`` is related. We could combine
+them together using :func:`.column_property`::
+
+ class A(Base):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+
+ class B(A):
+ __tablename__ = 'b'
+
+ # probably not what you want, but this is a demonstration
+ id = column_property(Column(Integer, primary_key=True), A.id)
+ a_id = Column(Integer, ForeignKey('a.id'))
+
+
+
I'm using Declarative and setting primaryjoin/secondaryjoin using an ``and_()`` or ``or_()``, and I am getting an error message about foreign keys.
------------------------------------------------------------------------------------------------------------------------------------------------------------------
prop = self._props.get(key, None)
if isinstance(prop, properties.ColumnProperty):
- if prop.parent is self:
- raise sa_exc.InvalidRequestError(
- "Implicitly combining column %s with column "
- "%s under attribute '%s'. Please configure one "
- "or more attributes for these same-named columns "
- "explicitly."
- % (prop.columns[-1], column, key))
+ if (
+ not self._inherits_equated_pairs or
+ (prop.columns[0], column) not in self._inherits_equated_pairs
+ ) and \
+ not prop.columns[0].shares_lineage(column) and \
+ prop.columns[0] is not self.version_id_col and \
+ column is not self.version_id_col:
+ warn_only = prop.parent is not self
+ msg = ("Implicitly combining column %s with column "
+ "%s under attribute '%s'. Please configure one "
+ "or more attributes for these same-named columns "
+ "explicitly." % (prop.columns[-1], column, key))
+ if warn_only:
+ util.warn(msg)
+ else:
+ raise sa_exc.InvalidRequestError(msg)
# existing properties.ColumnProperty from an inheriting
# mapper. make a copy and append our column to it
class Bar(Foo):
__tablename__ = 'bar'
- id = Column('id', Integer, primary_key=True)
+ bar_id = Column('id', Integer, primary_key=True)
foo_id = Column('foo_id', Integer)
__mapper_args__ = {'inherit_condition': foo_id == Foo.id}
bar = Table('bar', metadata,
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
- Column('data', String(20)))
+ Column('bar_data', String(20)))
blub = Table('blub', metadata,
Column('id', Integer, ForeignKey('bar.id'), primary_key=True),
Column('foo_id', Integer, ForeignKey('foo.id'), nullable=False),
- Column('data', String(20)))
+ Column('blub_data', String(20)))
def test_basic(self):
class Foo(object):
Column('x', String(10)),
Column('q', String(10)))
t2 = Table('t2', metadata,
- Column('id', Integer, primary_key=True,
+ Column('t2id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('y', String(10)),
Column('xid', ForeignKey('t1.id')))
polymorphic_identity='a')
mapper(B, table_b, inherits=A,
polymorphic_on=table_b.c.class_name,
- polymorphic_identity='b')
+ polymorphic_identity='b',
+ properties=dict(class_name=[table_a.c.class_name, table_b.c.class_name]))
mapper(C, table_c, inherits=B,
polymorphic_identity='c')
mapper(D, inherits=B,
bar = Table('bar', metadata,
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
- Column('data', String(20)))
+ Column('bar_data', String(20)))
blub = Table('blub', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('blub_id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('foo_id', Integer, ForeignKey('foo.id')),
Column('bar_id', Integer, ForeignKey('bar.id')),
- Column('data', String(20)))
+ Column('blub_data', String(20)))
@classmethod
def setup_classes(cls):
Column('data', String(30)))
bar = Table('bar', metadata,
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
- Column('data', String(30)))
+ Column('bar_data', String(30)))
bar_foo = Table('bar_foo', metadata,
Column('bar_id', Integer, ForeignKey('bar.id')),
A, B, C = cls.classes.A, cls.classes.B, cls.classes.C
mapper(A, cls.tables.a)
mapper(B, cls.tables.b, inherits=A,
- inherit_condition=cls.tables.a.c.id == cls.tables.b.c.id)
+ inherit_condition=cls.tables.a.c.id == cls.tables.b.c.id,
+ inherit_foreign_keys=cls.tables.b.c.id)
mapper(C, cls.tables.c, inherits=A,
- inherit_condition=cls.tables.a.c.id == cls.tables.c.c.id)
+ inherit_condition=cls.tables.a.c.id == cls.tables.c.c.id,
+ inherit_foreign_keys=cls.tables.c.c.id)
def test_ordering(self):
B, C = self.classes.B, self.classes.C
)
employee_table = Table("employees", metadata,
- Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
+ Column("eid", Integer, primary_key=True, test_needs_autoincrement=True),
Column("salary", Integer),
Column("person_id", Integer, ForeignKey("persons.id")),
)
person_mapper = mapper(Person, person_table)
mapper(Employee, employee_table, inherits=person_mapper,
properties={'pid':person_table.c.id,
- 'eid':employee_table.c.id})
+ 'eid':employee_table.c.eid})
self._do_test(False)
def test_explicit_composite_pk(self):
person_mapper = mapper(Person, person_table)
mapper(Employee, employee_table,
inherits=person_mapper,
- primary_key=[person_table.c.id, employee_table.c.id])
+ properties=dict(id=[employee_table.c.eid, person_table.c.id]),
+ primary_key=[person_table.c.id, employee_table.c.eid])
assert_raises_message(sa_exc.SAWarning,
r"On mapper Mapper\|Employee\|employees, "
"primary key column 'persons.id' is being "
- "combined with distinct primary key column 'employees.id' "
+ "combined with distinct primary key column 'employees.eid' "
"in attribute 'id'. Use explicit properties to give "
"each column its own mapped attribute name.",
self._do_test, True
@classmethod
def define_tables(cls, metadata):
- global base, subtable
+ global base, subtable, subtable_two
base = Table('base', metadata,
Column('base_id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('base_id', Integer, ForeignKey('base.base_id'), primary_key=True),
Column('subdata', String(255))
)
+ subtable_two = Table('subtable_two', metadata,
+ Column('base_id', Integer, primary_key=True),
+ Column('fk_base_id', Integer, ForeignKey('base.base_id')),
+ Column('subdata', String(255))
+ )
+
def test_plain(self):
# control case
# an exception in 0.7 due to the implicit conflict.
assert_raises(sa_exc.InvalidRequestError, go)
+ def test_pk_fk_different(self):
+ class Base(object):
+ pass
+ class Sub(Base):
+ pass
+
+ mapper(Base, base)
+
+ def go():
+ mapper(Sub, subtable_two, inherits=Base)
+ assert_raises_message(
+ sa_exc.SAWarning,
+ "Implicitly combining column base.base_id with "
+ "column subtable_two.base_id under attribute 'base_id'",
+ go
+ )
+
def test_plain_descriptor(self):
"""test that descriptors prevent inheritance from propigating properties to subclasses."""
Table('sub', metadata,
Column('id', Integer, ForeignKey('base.id'), primary_key=True),
Column('sub', String(50)),
- Column('counter', Integer, server_default="1"),
- Column('counter2', Integer, server_default="1")
+ Column('subcounter', Integer, server_default="1"),
+ Column('subcounter2', Integer, server_default="1")
)
Table('subsub', metadata,
Column('id', Integer, ForeignKey('sub.id'), primary_key=True),
- Column('counter2', Integer, server_default="1")
+ Column('subsubcounter2', Integer, server_default="1")
)
Table('with_comp', metadata,
Column('id', Integer, ForeignKey('base.id'), primary_key=True),
mapper(Base, base)
mapper(JoinBase, base.outerjoin(sub), properties=util.OrderedDict(
[('id', [base.c.id, sub.c.id]),
- ('counter', [base.c.counter, sub.c.counter])])
+ ('counter', [base.c.counter, sub.c.subcounter])])
)
mapper(SubJoinBase, inherits=JoinBase)
go,
CompiledSQL(
"SELECT base.id AS base_id, sub.id AS sub_id, "
- "base.counter AS base_counter, sub.counter AS sub_counter, "
- "base.data AS base_data, "
- "base.type AS base_type, sub.sub AS sub_sub, "
- "sub.counter2 AS sub_counter2 FROM base "
- "LEFT OUTER JOIN sub ON base.id = sub.id "
+ "base.counter AS base_counter, sub.subcounter AS sub_subcounter, "
+ "base.data AS base_data, base.type AS base_type, "
+ "sub.sub AS sub_sub, sub.subcounter2 AS sub_subcounter2 "
+ "FROM base LEFT OUTER JOIN sub ON base.id = sub.id "
"WHERE base.id = :param_1",
{'param_1': sjb_id}
),
),
)
def go():
- eq_( s1.counter2, 1 )
+ eq_( s1.subcounter2, 1 )
self.assert_sql_execution(
testing.db,
go,
CompiledSQL(
- "SELECT sub.counter AS sub_counter, base.counter AS base_counter, "
- "sub.counter2 AS sub_counter2 FROM base JOIN sub ON "
- "base.id = sub.id WHERE base.id = :param_1",
+ "SELECT base.counter AS base_counter, sub.subcounter AS sub_subcounter, "
+ "sub.subcounter2 AS sub_subcounter2 FROM base JOIN sub "
+ "ON base.id = sub.id WHERE base.id = :param_1",
lambda ctx:{'param_1': s1.id}
),
)
s1 = Sub()
assert m._optimized_get_statement(attributes.instance_state(s1),
- ['counter2']) is None
+ ['subcounter2']) is None
# loads s1.id as None
eq_(s1.id, None)
# this now will come up with a value of None for id - should reject
assert m._optimized_get_statement(attributes.instance_state(s1),
- ['counter2']) is None
+ ['subcounter2']) is None
s1.id = 1
attributes.instance_state(s1)._commit_all(s1.__dict__, None)
assert m._optimized_get_statement(attributes.instance_state(s1),
- ['counter2']) is not None
+ ['subcounter2']) is not None
def test_load_expired_on_pending_twolevel(self):
base, sub, subsub = (self.tables.base,
mapper(Sub, sub, inherits=Base, polymorphic_identity='sub')
mapper(SubSub, subsub, inherits=Sub, polymorphic_identity='subsub')
sess = Session()
- s1 = SubSub(data='s1', counter=1)
+ s1 = SubSub(data='s1', counter=1, subcounter=2)
sess.add(s1)
self.assert_sql_execution(
testing.db,
[{'data':'s1','type':'subsub','counter':1}]
),
CompiledSQL(
- "INSERT INTO sub (id, sub, counter) VALUES "
- "(:id, :sub, :counter)",
- lambda ctx:[{'counter': 1, 'sub': None, 'id': s1.id}]
+ "INSERT INTO sub (id, sub, subcounter) VALUES "
+ "(:id, :sub, :subcounter)",
+ lambda ctx:[{'subcounter': 2, 'sub': None, 'id': s1.id}]
),
CompiledSQL(
"INSERT INTO subsub (id) VALUES (:id)",
def go():
eq_(
- s1.counter2, 1
+ s1.subcounter2, 1
)
self.assert_sql_execution(
testing.db,
go,
CompiledSQL(
- "SELECT subsub.counter2 AS subsub_counter2, "
- "sub.counter2 AS sub_counter2 FROM subsub, sub "
+ "SELECT subsub.subsubcounter2 AS subsub_subsubcounter2, "
+ "sub.subcounter2 AS sub_subcounter2 FROM subsub, sub "
"WHERE :param_1 = sub.id AND sub.id = subsub.id",
lambda ctx:{'param_1': s1.id}
),
bar = Table('bar', metadata,
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
- Column('data', String(20)))
+ Column('bar_data', String(20)))
blub = Table('blub', metadata,
Column('id', Integer, ForeignKey('bar.id'), primary_key=True),
- Column('data', String(20)))
+ Column('blub_data', String(20)))
bar_foo = Table('bar_foo', metadata,
Column('bar_id', Integer, ForeignKey('bar.id')),
def define_tables(cls, metadata):
Table('a', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('data', String(30)),
Column('cid', Integer, ForeignKey('c.id')))
Table('b', metadata,
Column('id', Integer, ForeignKey("a.id"), primary_key=True),
- Column('data', String(30)))
+ )
Table('c', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('data', String(30)),
Column('aid', Integer,
ForeignKey('a.id', use_alter=True, name="foo")))
pass
mapper(A, users)
- mapper(B, addresses, inherits=A)
+ mapper(B, addresses, inherits=A,
+ properties={'address_id': addresses.c.id})
def init_a(target, args, kwargs):
canary.append(('init_a', target))
pass
mapper(User, users)
- mapper(AdminUser, addresses, inherits=User)
+ mapper(AdminUser, addresses, inherits=User,
+ properties={'address_id': addresses.c.id})
canary1 = self.listen_all(User, propagate=True)
canary2 = self.listen_all(User)
class AdminUser(User):
pass
- mapper(AdminUser, addresses, inherits=User)
+ mapper(AdminUser, addresses, inherits=User,
+ properties={'address_id': addresses.c.id})
canary3 = self.listen_all(AdminUser)
sess = create_session()
pass
mapper(User, users)
- mapper(AdminUser, addresses, inherits=User)
+ mapper(AdminUser, addresses, inherits=User,
+ properties={'address_id': addresses.c.id})
fn = Mock()
event.listen(User.name, "set", fn, propagate=True)
pass
mapper(User, users, extension=Ext())
- mapper(AdminUser, addresses, inherits=User)
+ mapper(AdminUser, addresses, inherits=User,
+ properties={'address_id': addresses.c.id})
sess = create_session()
am = AdminUser(name='au1', email_address='au1@e1')
ext = Ext()
mapper(User, users, extension=ext)
- mapper(AdminUser, addresses, inherits=User, extension=ext)
+ mapper(AdminUser, addresses, inherits=User, extension=ext,
+ properties={'address_id': addresses.c.id})
sess = create_session()
am = AdminUser(name="au1", email_address="au1@e1")
user_table = self.tables.users
addresses_table = self.tables.addresses
mapper(Foo, user_table, with_polymorphic=(Bar,))
- mapper(Bar, addresses_table, inherits=Foo)
+ mapper(Bar, addresses_table, inherits=Foo, properties={
+ 'address_id': addresses_table.c.id
+ })
i1 = inspect(Foo)
i2 = inspect(Foo)
assert i1.selectable is i2.selectable
class Foo(User):pass
mapper(User, users)
- mapper(Foo, addresses, inherits=User)
+ mapper(Foo, addresses, inherits=User, properties={
+ 'address_id': addresses.c.id
+ })
assert getattr(Foo().__class__, 'name').impl is not None
def test_deferred_subclass_attribute_instrument(self):
class Foo(User):pass
mapper(User, users)
configure_mappers()
- mapper(Foo, addresses, inherits=User)
+ mapper(Foo, addresses, inherits=User, properties={
+ 'address_id': addresses.c.id
+ })
assert getattr(Foo().__class__, 'name').impl is not None
def test_check_descriptor_as_method(self):
class SubUser(User):
pass
m = mapper(User, users)
- m2 = mapper(SubUser, addresses, inherits=User)
+ m2 = mapper(SubUser, addresses, inherits=User, properties={
+ 'address_id': addresses.c.id
+ })
m3 = mapper(Address, addresses, properties={
'foo':relationship(m2)
})
pass
m1 = mapper(User, users, polymorphic_identity='user')
m2 = mapper(AddressUser, addresses, inherits=User,
- polymorphic_identity='address')
+ polymorphic_identity='address', properties={
+ 'address_id': addresses.c.id
+ })
m3 = mapper(AddressUser, addresses, non_primary=True)
assert m3._identity_class is m2._identity_class
eq_(
Table("a", metadata,
Column('aid', Integer, primary_key=True,
test_needs_autoincrement=True),
- Column('data', String(30)))
+ Column('adata', String(30)))
Table("b", metadata,
Column('bid', Integer, primary_key=True,
test_needs_autoincrement=True),
Column("a_id", Integer, ForeignKey("a.aid")),
- Column('data', String(30)))
+ Column('bdata', String(30)))
Table("c", metadata,
Column('cid', Integer, primary_key=True,
test_needs_autoincrement=True),
Column("b_id", Integer, ForeignKey("b.bid")),
- Column('data', String(30)))
+ Column('cdata', String(30)))
Table("d", metadata,
Column('did', Integer, primary_key=True,
test_needs_autoincrement=True),
Column("a_id", Integer, ForeignKey("a.aid")),
- Column('data', String(30)))
+ Column('ddata', String(30)))
def test_o2m_oncascade(self):
a, c, b = (self.tables.a,
# define a mapper for AddressUser that inherits the User.mapper, and
# joins on the id column
- mapper(AddressUser, addresses, inherits=m1)
+ mapper(AddressUser, addresses, inherits=m1, properties={
+ 'address_id': addresses.c.id
+ })
au = AddressUser(name='u', email_address='u@e')
@classmethod
def define_tables(cls, metadata):
Table('parent', metadata,
- Column('id', Integer, primary_key=True),
+ Column('pid', Integer, primary_key=True),
Column('pdata', String(30))
)
Table('child', metadata,
- Column('id', Integer, primary_key=True),
- Column('pid', Integer, ForeignKey('parent.id')),
+ Column('cid', Integer, primary_key=True),
+ Column('pid', Integer, ForeignKey('parent.pid')),
Column('cdata', String(30))
)
mapper(C, child, inherits=P)
sess = create_session()
- c1 = C(id=1, pdata='c1', cdata='c1')
+ c1 = C(pid=1, cid=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')
+ c2 = C(pid=1, cid=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}
+ CompiledSQL("UPDATE parent SET pdata=:pdata WHERE parent.pid = :parent_pid",
+ {'pdata':'c2', 'parent_pid':1}
),
# this fires as of [ticket:1362], since we synchronzize
# PK/FKs on UPDATES. c2 is new so the history shows up as
# pure added, update occurs. If a future change limits the
# sync operation during _save_obj().update, this is safe to remove again.
- CompiledSQL("UPDATE child SET pid=:pid WHERE child.id = :child_id",
- {'pid':1, 'child_id':1}
+ CompiledSQL("UPDATE child SET pid=:pid WHERE child.cid = :child_cid",
+ {'pid':1, 'child_cid':1}
)
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
