conflicts with a subquery or column of the same name
on the parent object. [ticket:1019]
+- declarative extension
+ - Joined table inheritance mappers use a slightly relaxed
+ function to create the "inherit condition" to the parent
+ table, so that other foreign keys to not-yet-declared
+ Table objects don't trigger an error.
+
- sql
- Added COLLATE support via the .collate(<collation>)
expression operator and collate(<expr>, <collation>) sql
class NoSuchColumnError(KeyError, SQLAlchemyError):
"""Raised by ``RowProxy`` when a nonexistent column is requested from a row."""
-
+
+class NoSuchTableError(InvalidRequestError):
+ """Raised by ``ForeignKey`` when the referred ``Table`` cannot be located."""
class DisconnectionError(SQLAlchemyError):
"""Raised within ``Pool`` when a disconnect is detected on a raw DB-API connection.
from sqlalchemy.orm.interfaces import MapperProperty
from sqlalchemy.orm.properties import PropertyLoader, ColumnProperty
from sqlalchemy import util, exceptions
+from sqlalchemy.sql import util as sql_util
__all__ = ['declarative_base', 'synonym_for', 'comparable_using',
if 'inherits' not in mapper_args:
inherits = cls.__mro__[1]
inherits = cls._decl_class_registry.get(inherits.__name__, None)
- mapper_args['inherits'] = inherits
+ if inherits:
+ mapper_args['inherits'] = inherits
+ if not mapper_args.get('concrete', False) and table:
+ # figure out the inherit condition with relaxed rules about nonexistent tables,
+ # to allow for ForeignKeys to not-yet-defined tables (since we know for sure that our parent
+ # table is defined within the same MetaData)
+ mapper_args['inherit_condition'] = sql_util.join_condition(inherits.__table__, table, ignore_nonexistent_tables=True)
if hasattr(cls, '__mapper_cls__'):
mapper_cls = util.unbound_method_to_callable(cls.__mapper_cls__)
# figure out inherit condition from our table to the immediate table
# of the inherited mapper, not its full table which could pull in other
# stuff we dont want (allows test/inheritance.InheritTest4 to pass)
- self.inherit_condition = sql.join(self.inherits.local_table, self.local_table).onclause
+ self.inherit_condition = sqlutil.join_condition(self.inherits.local_table, self.local_table)
self.mapped_table = sql.join(self.inherits.mapped_table, self.local_table, self.inherit_condition)
fks = util.to_set(self.inherit_foreign_keys)
else:
(schema,tname,colname) = m.group(1,2,3)
if _get_table_key(tname, schema) not in parenttable.metadata:
- raise exceptions.InvalidRequestError(
+ raise exceptions.NoSuchTableError(
"Could not find table '%s' with which to generate a "
"foreign key" % tname)
table = Table(tname, parenttable.metadata,
return self.left, self.right, self.onclause
def __match_primaries(self, primary, secondary):
- crit = []
- constraints = util.Set()
- for fk in secondary.foreign_keys:
- col = fk.get_referent(primary)
- if col:
- crit.append(col == fk.parent)
- constraints.add(fk.constraint)
- if primary is not secondary:
- for fk in primary.foreign_keys:
- col = fk.get_referent(secondary)
- if col:
- crit.append(col == fk.parent)
- constraints.add(fk.constraint)
- if len(crit) == 0:
- raise exceptions.ArgumentError(
- "Can't find any foreign key relationships "
- "between '%s' and '%s'" % (primary.description, secondary.description))
- elif len(constraints) > 1:
- raise exceptions.ArgumentError(
- "Can't determine join between '%s' and '%s'; "
- "tables have more than one foreign key "
- "constraint relationship between them. "
- "Please specify the 'onclause' of this "
- "join explicitly." % (primary.description, secondary.description))
- elif len(crit) == 1:
- return (crit[0])
- else:
- return and_(*crit)
+ global sql_util
+ if not sql_util:
+ from sqlalchemy.sql import util as sql_util
+ return sql_util.join_condition(primary, secondary)
def select(self, whereclause=None, fold_equivalents=False, **kwargs):
"""Create a ``Select`` from this ``Join``.
visitors.traverse(clause, visit_column=visit_column)
return cols
-
+def join_condition(a, b, ignore_nonexistent_tables=False):
+ """create a join condition between two tables.
+
+ ignore_nonexistent_tables=True allows a join condition to be
+ determined between two tables which may contain references to
+ other not-yet-defined tables. In general the NoSuchTableError
+ raised is only required if the user is trying to join selectables
+ across multiple MetaData objects (which is an extremely rare use
+ case).
+
+ """
+ crit = []
+ constraints = util.Set()
+ for fk in b.foreign_keys:
+ try:
+ col = fk.get_referent(a)
+ except exceptions.NoSuchTableError:
+ if ignore_nonexistent_tables:
+ continue
+ else:
+ raise
+
+ if col:
+ crit.append(col == fk.parent)
+ constraints.add(fk.constraint)
+
+ if a is not b:
+ for fk in a.foreign_keys:
+ try:
+ col = fk.get_referent(b)
+ except exceptions.NoSuchTableError:
+ if ignore_nonexistent_tables:
+ continue
+ else:
+ raise
+
+ if col:
+ crit.append(col == fk.parent)
+ constraints.add(fk.constraint)
+
+ if len(crit) == 0:
+ raise exceptions.ArgumentError(
+ "Can't find any foreign key relationships "
+ "between '%s' and '%s'" % (a.description, b.description))
+ elif len(constraints) > 1:
+ raise exceptions.ArgumentError(
+ "Can't determine join between '%s' and '%s'; "
+ "tables have more than one foreign key "
+ "constraint relationship between them. "
+ "Please specify the 'onclause' of this "
+ "join explicitly." % (a.description, b.description))
+ elif len(crit) == 1:
+ return (crit[0])
+ else:
+ return and_(*crit)
+
+
def reduce_columns(columns, *clauses):
"""given a list of columns, return a 'reduced' set based on natural equivalents.
sess.clear()
self.assertEquals(sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).first(), c2)
-
+
+ def test_inheritance_with_undefined_relation(self):
+ Base = declarative_base()
+
+ class Parent(Base):
+ __tablename__ = 'parent'
+ id = Column('id', Integer, primary_key=True)
+ tp = Column('type', String(50))
+ __mapper_args__ = dict(polymorphic_on = tp)
+
+
+ class Child1(Parent):
+ __tablename__ = 'child1'
+ id = Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
+ related_child2 = Column('c2', Integer, ForeignKey('child2.id'))
+ __mapper_args__ = dict(polymorphic_identity = 'child1')
+
+ # no exception is raised by the ForeignKey to "child2" even though child2 doesn't exist yet
+
+ class Child2(Parent):
+ __tablename__ = 'child2'
+ id = Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
+ related_child1 = Column('c1', Integer)
+ __mapper_args__ = dict(polymorphic_identity = 'child2')
+
+ compile_mappers() # no exceptions here
+
def test_relation_reference(self):
class Address(Base, Fixture):
__tablename__ = 'addresses'
from sqlalchemy import *
from testlib import *
from sqlalchemy.sql import util as sql_util
+from sqlalchemy import exceptions
metadata = MetaData()
table = Table('table1', metadata,
assert u.corresponding_column(s.oid_column) is u.oid_column
assert u.corresponding_column(s2.oid_column) is u.oid_column
+ def test_two_metadata_join_raises(self):
+ m = MetaData()
+ m2 = MetaData()
+
+ t1 = Table('t1', m, Column('id', Integer), Column('id2', Integer))
+ t2 = Table('t2', m, Column('id', Integer, ForeignKey('t1.id')))
+ t3 = Table('t3', m2, Column('id', Integer, ForeignKey('t1.id2')))
+
+ s = select([t2, t3], use_labels=True)
+
+ self.assertRaises(exceptions.NoSuchTableError, s.join, t1)
+
class PrimaryKeyTest(TestBase, AssertsExecutionResults):
def test_join_pk_collapse_implicit(self):
"""test that redundant columns in a join get 'collapsed' into a minimal primary key,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
