return None
@classmethod
- def _listen(cls, target, identifier, fn, propagate=False):
- getattr(target.dispatch, identifier).append(fn, target, propagate)
+ def _listen(cls, target, identifier, fn, propagate=False, insert=False):
+ if insert:
+ getattr(target.dispatch, identifier).insert(fn, target, propagate)
+ else:
+ getattr(target.dispatch, identifier).append(fn, target, propagate)
@classmethod
def _remove(cls, target, identifier, fn):
self.__doc__ = fn.__doc__
self._clslevel = util.defaultdict(list)
+ def insert(self, obj, target, propagate):
+ assert isinstance(target, type), \
+ "Class-level Event targets must be classes."
+
+ stack = [target]
+ while stack:
+ cls = stack.pop(0)
+ stack.extend(cls.__subclasses__())
+ self._clslevel[cls].insert(0, obj)
+
def append(self, obj, target, propagate):
assert isinstance(target, type), \
"Class-level Event targets must be classes."
and not only_propagate or l in self.propagate
])
+ def insert(self, obj, target, propagate):
+ if obj not in self.listeners:
+ self.listeners.insert(0, obj)
+ if propagate:
+ self.propagate.add(obj)
+
def append(self, obj, target, propagate):
if obj not in self.listeners:
self.listeners.append(obj)
# allow user-overrides
self._init_items(*args)
+ @property
+ def _sorted_constraints(self):
+ """Return the set of constraints as a list, sorted by creation order."""
+
+ return sorted(self.constraints, key=lambda c:c._creation_order)
+
def _init_existing(self, *args, **kwargs):
autoload = kwargs.pop('autoload', False)
autoload_with = kwargs.pop('autoload_with', None)
self.deferrable = deferrable
self.initially = initially
self._create_rule = _create_rule
+ util.set_creation_order(self)
@property
def table(self):
if table.primary_key:
constraints.append(table.primary_key)
- constraints.extend([c for c in table.constraints
+ constraints.extend([c for c in table._sorted_constraints
if c is not table.primary_key])
return ", \n\t".join(p for p in
eq_(len(Target().dispatch.event_one), 2)
eq_(len(t1.dispatch.event_one), 3)
+ def test_append_vs_insert(self):
+ def listen_one(x, y):
+ pass
+
+ def listen_two(x, y):
+ pass
+
+ def listen_three(x, y):
+ pass
+
+ event.listen(Target, "event_one", listen_one)
+ event.listen(Target, "event_one", listen_two)
+ event.listen(Target, "event_one", listen_three, insert=True)
+
+ eq_(
+ list(Target().dispatch.event_one),
+ [listen_three, listen_one, listen_two]
+ )
+
class TestAcceptTargets(TestBase):
"""Test default target acceptance."""
factory = lambda **kw: CheckConstraint('a < b', **kw)
self._test_deferrable(factory)
+ def test_multiple(self):
+ m = MetaData()
+ foo = Table("foo", m,
+ Column('id', Integer, primary_key=True),
+ Column('bar', Integer, primary_key=True)
+ )
+ tb = Table("some_table", m,
+ Column('id', Integer, primary_key=True),
+ Column('foo_id', Integer, ForeignKey('foo.id')),
+ Column('foo_bar', Integer, ForeignKey('foo.bar')),
+ )
+ self.assert_compile(
+ schema.CreateTable(tb),
+ "CREATE TABLE some_table ("
+ "id INTEGER NOT NULL, "
+ "foo_id INTEGER, "
+ "foo_bar INTEGER, "
+ "PRIMARY KEY (id), "
+ "FOREIGN KEY(foo_id) REFERENCES foo (id), "
+ "FOREIGN KEY(foo_bar) REFERENCES foo (bar))"
+ )
+
def test_deferrable_column_check(self):
t = Table('tbl', MetaData(),
Column('a', Integer),