- mapper() will ignore non-configured foreign keys
to unrelated tables when determining inherit
- condition between parent and child class.
- This is equivalent to behavior already
- applied to declarative. [ticket:2153]
- Also in 0.6.8.
+ condition between parent and child class,
+ but will raise as usual for unresolved
+ columns and table names regarding the inherited
+ table. This is an enhanced generalization of
+ behavior that was already applied to declarative
+ previously. [ticket:2153] 0.6.8 has a more
+ conservative version of this which doesn't
+ fundamentally alter how join conditions
+ are determined.
- It is an error to call query.get() when the
given entity is not a single, full class
Also in 0.6.8.
- sql
+ - Changed the handling in determination of join
+ conditions such that foreign key errors are
+ only considered between the two given tables.
+ That is, t1.join(t2) will report FK errors
+ that involve 't1' or 't2', but anything
+ involving 't3' will be skipped. This affects
+ join(), as well as ORM relationship and
+ inherit condition logic.
+
- Some improvements to error handling inside
of the execute procedure to ensure auto-close
connections are really closed when very
class NoReferencedTableError(NoReferenceError):
"""Raised by ``ForeignKey`` when the referred ``Table`` cannot be located."""
+ def __init__(self, message, tname):
+ super(NoReferencedTableError, self).__init__(message)
+ self.table_name = tname
+
class NoReferencedColumnError(NoReferenceError):
"""Raised by ``ForeignKey`` when the referred ``Column`` cannot be located."""
+ def __init__(self, message, tname, cname):
+ super(NoReferencedColumnError, self).__init__(message)
+ self.table_name = tname
+ self.column_name = cname
+
class NoSuchTableError(InvalidRequestError):
"""Table does not exist or is not visible to a connection."""
# want (allows test/inheritance.InheritTest4 to pass)
self.inherit_condition = sqlutil.join_condition(
self.inherits.local_table,
- self.local_table,
- ignore_nonexistent_tables=True)
+ self.local_table)
self.mapped_table = sql.join(
self.inherits.mapped_table,
self.local_table,
raise exc.NoReferencedTableError(
"Foreign key associated with column '%s' could not find "
"table '%s' with which to generate a "
- "foreign key to target column '%s'" % (self.parent, tname, colname))
+ "foreign key to target column '%s'" % (self.parent, tname, colname),
+ tname)
table = Table(tname, parenttable.metadata,
mustexist=True, schema=schema)
raise exc.NoReferencedColumnError(
"Could not create ForeignKey '%s' on table '%s': "
"table '%s' has no column named '%s'" % (
- self._colspec, parenttable.name, table.name, key))
+ self._colspec, parenttable.name, table.name, key),
+ table.name, key)
elif hasattr(self._colspec, '__clause_element__'):
_column = self._colspec.__clause_element__()
return visitors.cloned_traverse(crit, {}, {'binary':visit_binary})
-
def join_condition(a, b, ignore_nonexistent_tables=False, a_subset=None):
"""create a join condition between two tables or selectables.
between the two selectables. If there are multiple ways
to join, or no way to join, an error is raised.
- :param ignore_nonexistent_tables: This flag will cause the
- function to silently skip over foreign key resolution errors
- due to nonexistent tables - the assumption is that these
- tables have not yet been defined within an initialization process
- and are not significant to the operation.
+ :param ignore_nonexistent_tables: Deprecated - this
+ flag is no longer used. Only resolution errors regarding
+ the two given tables are propagated.
:param a_subset: An optional expression that is a sub-component
of ``a``. An attempt will be made to join to just this sub-component
key=lambda fk:fk.parent._creation_order):
try:
col = fk.get_referent(left)
- except exc.NoReferencedTableError:
- if ignore_nonexistent_tables:
- continue
- else:
+ except exc.NoReferenceError, nrte:
+ if nrte.table_name == left.name:
raise
+ else:
+ continue
if col is not None:
crit.append(col == fk.parent)
key=lambda fk:fk.parent._creation_order):
try:
col = fk.get_referent(b)
- except exc.NoReferencedTableError:
- if ignore_nonexistent_tables:
- continue
- else:
+ except exc.NoReferenceError, nrte:
+ if nrte.table_name == b.name:
raise
+ else:
+ # this is totally covered. can't get
+ # coverage to mark it.
+ continue
if col is not None:
crit.append(col == fk.parent)
eq_(mc.primary_key, (parent.c.id,))
class InhCondTest(fixtures.TestBase):
- def test_inh_cond_ignores_others(self):
+ def test_inh_cond_nonexistent_table_unrelated(self):
metadata = MetaData()
base_table = Table("base", metadata,
Column("id", Integer, primary_key=True)
base_table.c.id==derived_table.c.id
)
- def test_inh_cond_fails_notfound(self):
+ def test_inh_cond_nonexistent_col_unrelated(self):
+ m = MetaData()
+ base_table = Table("base", m,
+ Column("id", Integer, primary_key=True)
+ )
+ derived_table = Table("derived", m,
+ Column("id", Integer, ForeignKey('base.id'),
+ primary_key=True),
+ Column('order_id', Integer, ForeignKey('order.foo'))
+ )
+ order_table = Table('order', m, Column('id', Integer, primary_key=True))
+ class Base(object):
+ pass
+
+ class Derived(Base):
+ pass
+
+ mapper(Base, base_table)
+
+ # succeeds, despite "order.foo" doesn't exist
+ m2 = mapper(Derived, derived_table, inherits=Base)
+ assert m2.inherit_condition.compare(
+ base_table.c.id==derived_table.c.id
+ )
+
+ def test_inh_cond_no_fk(self):
metadata = MetaData()
base_table = Table("base", metadata,
Column("id", Integer, primary_key=True)
Derived, derived_table, inherits=Base
)
- def test_inh_cond_fails_separate_metas(self):
+ def test_inh_cond_nonexistent_table_related(self):
m1 = MetaData()
m2 = MetaData()
base_table = Table("base", m1,
mapper(Base, base_table)
- # this used to be "can't resolve foreign key base.id",
- # but with the flag on, we just get "can't find". this is
- # the less-than-ideal case that prevented us from doing this
- # for mapper(), not just declarative, in the first place.
- # there is no case where the failure would be silent -
- # there is either a single join condition between the two tables
- # or there's not.
+ # the ForeignKey def is correct but there are two
+ # different metadatas. Would like the traditional
+ # "noreferencedtable" error to raise so that the
+ # user is directed towards the FK definition in question.
assert_raises_message(
- sa_exc.ArgumentError,
- "Can't find any foreign key relationships between "
- "'base' and 'derived'.",
+ sa_exc.NoReferencedTableError,
+ "Foreign key associated with column 'derived.id' "
+ "could not find table 'base' with which to generate "
+ "a foreign key to target column 'id'",
mapper,
Derived, derived_table, inherits=Base
)
+ def test_inh_cond_nonexistent_col_related(self):
+ m = MetaData()
+ base_table = Table("base", m,
+ Column("id", Integer, primary_key=True)
+ )
+ derived_table = Table("derived", m,
+ Column("id", Integer, ForeignKey('base.q'),
+ primary_key=True),
+ )
+
+ class Base(object):
+ pass
+
+ class Derived(Base):
+ pass
+
+ mapper(Base, base_table)
+
+ assert_raises_message(
+ sa_exc.NoReferencedColumnError,
+ "Could not create ForeignKey 'base.q' on table "
+ "'derived': table 'base' has no column named 'q'",
+ mapper,
+ Derived, derived_table, inherits=Base
+ )
+
+
class PKDiscriminatorTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
configure_mappers)
- def test_fk_error_raised(self):
+ def test_fk_error_not_raised_unrelated(self):
m = MetaData()
t1 = Table('t1', m,
Column('id', Integer, primary_key=True),
mapper(C1, t1, properties={'c2':relationship(C2)})
mapper(C2, t3)
-
- assert_raises(sa.exc.NoReferencedColumnError, configure_mappers)
+ assert C1.c2.property.primaryjoin.compare(t1.c.id==t3.c.t1id)
def test_join_error_raised(self):
m = MetaData()
assert_raises(exc.NoReferencedTableError, s.join, t1)
+class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults):
def test_join_condition(self):
m = MetaData()
t1t2.join, t2t3.select(use_labels=True))
+ def test_join_cond_no_such_unrelated_table(self):
+ m = MetaData()
+ # bounding the "good" column with two "bad" ones is so to
+ # try to get coverage to get the "continue" statements
+ # in the loop...
+ t1 = Table('t1', m,
+ Column('y', Integer, ForeignKey('t22.id')),
+ Column('x', Integer, ForeignKey('t2.id')),
+ Column('q', Integer, ForeignKey('t22.id')),
+ )
+ t2 = Table('t2', m, Column('id', Integer))
+ assert sql_util.join_condition(t1, t2).compare(t1.c.x==t2.c.id)
+ assert sql_util.join_condition(t2, t1).compare(t1.c.x==t2.c.id)
+
+ def test_join_cond_no_such_unrelated_column(self):
+ m = MetaData()
+ t1 = Table('t1', m, Column('x', Integer, ForeignKey('t2.id')),
+ Column('y', Integer, ForeignKey('t3.q')))
+ t2 = Table('t2', m, Column('id', Integer))
+ t3 = Table('t3', m, Column('id', Integer))
+ assert sql_util.join_condition(t1, t2).compare(t1.c.x==t2.c.id)
+ assert sql_util.join_condition(t2, t1).compare(t1.c.x==t2.c.id)
+
+ def test_join_cond_no_such_related_table(self):
+ m1 = MetaData()
+ m2 = MetaData()
+ t1 = Table('t1', m1, Column('x', Integer, ForeignKey('t2.id')))
+ t2 = Table('t2', m2, Column('id', Integer))
+ assert_raises_message(
+ exc.NoReferencedTableError,
+ "Foreign key associated with column 't1.x' could not find "
+ "table 't2' with which to generate a foreign key to "
+ "target column 'id'",
+ sql_util.join_condition, t1, t2
+ )
+
+ assert_raises_message(
+ exc.NoReferencedTableError,
+ "Foreign key associated with column 't1.x' could not find "
+ "table 't2' with which to generate a foreign key to "
+ "target column 'id'",
+ sql_util.join_condition, t2, t1
+ )
+
+ def test_join_cond_no_such_related_column(self):
+ m = MetaData()
+ t1 = Table('t1', m, Column('x', Integer, ForeignKey('t2.q')))
+ t2 = Table('t2', m, Column('id', Integer))
+ assert_raises_message(
+ exc.NoReferencedColumnError,
+ "Could not create ForeignKey 't2.q' on table 't1': table 't2' has no column named 'q'",
+ sql_util.join_condition, t1, t2
+ )
+
+ assert_raises_message(
+ exc.NoReferencedColumnError,
+ "Could not create ForeignKey 't2.q' on table 't1': table 't2' has no column named 'q'",
+ sql_util.join_condition, t2, t1
+ )
+
class PrimaryKeyTest(fixtures.TestBase, AssertsExecutionResults):
def test_join_pk_collapse_implicit(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
