'duplicate' columns from the resulting column clause that are known to be
equivalent based on the join condition. this is of great usage when
constructing subqueries of joins which Postgres complains about if
- duplicate column names are present.
+ duplicate column names are present.
+ - fixed use_alter flag on ForeignKeyConstraint [ticket:503]
- orm:
- a full select() construct can be passed to query.select() (which
worked anyway), but also query.selectfirst(), query.selectone() which
pass
+class CircularDependencyError(SQLAlchemyError):
+ """Raised by topological sorts when a circular dependency is detected"""
+ pass
+
class FlushError(SQLAlchemyError):
"""Raised when an invalid condition is detected upon a ``flush()``."""
pass
visitor.visit_foreign_key_constraint(self)
def append_element(self, col, refcol):
- fk = ForeignKey(refcol, constraint=self)
+ fk = ForeignKey(refcol, constraint=self, name=self.name, onupdate=self.onupdate, ondelete=self.ondelete, use_alter=self.use_alter)
fk._set_parent(self.table.c[col])
self._append_fk(fk)
self.elements.add(fk)
def copy(self):
- return ForeignKeyConstraint([x.parent.name for x in self.elements], [x._get_colspec() for x in self.elements], name=self.name, onupdate=self.onupdate, ondelete=self.ondelete)
+ return ForeignKeyConstraint([x.parent.name for x in self.elements], [x._get_colspec() for x in self.elements], name=self.name, onupdate=self.onupdate, ondelete=self.ondelete, use_alter=self.use_alter)
class PrimaryKeyConstraint(Constraint):
def __init__(self, *columns, **kwargs):
import string, StringIO
from sqlalchemy import util
-from sqlalchemy.exceptions import *
+from sqlalchemy.exceptions import CircularDependencyError
class _Node(object):
"""Represent each item in the sort.
n.cycles = util.Set([n])
continue
else:
- raise FlushError("Self-referential dependency detected " + repr(t))
+ raise CircularDependencyError("Self-referential dependency detected " + repr(t))
childnode = nodes[t[1]]
parentnode = nodes[t[0]]
edges.add((parentnode, childnode))
continue
else:
# long cycles not allowed
- raise FlushError("Circular dependency detected " + repr(edges) + repr(queue))
+ raise CircularDependencyError("Circular dependency detected " + repr(edges) + repr(queue))
node = queue.pop()
if not hasattr(node, '_cyclical'):
output.append(node)
)
metadata.create_all()
+ def test_circular_constraint(self):
+ a = Table("a", metadata,
+ Column('id', Integer, primary_key=True),
+ Column('bid', Integer),
+ ForeignKeyConstraint(["bid"], ["b.id"], name="afk")
+ )
+ b = Table("b", metadata,
+ Column('id', Integer, primary_key=True),
+ Column("aid", Integer),
+ ForeignKeyConstraint(["aid"], ["a.id"], use_alter=True, name="bfk")
+ )
+ metadata.create_all()
+
+ def test_circular_constraint_2(self):
+ a = Table("a", metadata,
+ Column('id', Integer, primary_key=True),
+ Column('bid', Integer, ForeignKey("b.id")),
+ )
+ b = Table("b", metadata,
+ Column('id', Integer, primary_key=True),
+ Column("aid", Integer, ForeignKey("a.id", use_alter=True, name="bfk")),
+ )
+ metadata.create_all()
+
@testbase.unsupported('mysql')
def test_check_constraint(self):
foo = Table('foo', metadata,