Issue 1:
Migrated issue, originally created by Michael Bayer (@zzzeek)

these rely upon the sqlite_master view which does not take schema name into account:

--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -16,6 +16,7 @@ from sqlalchemy.schema import DDL, Index
from sqlalchemy import event
from sqlalchemy.sql.elements import quoted_name
from sqlalchemy import ForeignKey
+import re

metadata, users = None, None

@@ -760,6 +761,50 @@ class ComponentReflectionTest(fixtures.TablesTest):
eq_(names_that_duplicate_index, idx_names)
eq_(uq_names, set())

+ @testing.requires.check_constraint_reflection
+ def test_get_check_constraints(self):
+ self._test_get_check_constraints()
+
+ @testing.requires.check_constraint_reflection
+ @testing.requires.schemas
+ def test_get_check_constraints_schema(self):
+ self._test_get_check_constraints(schema=testing.config.test_schema)
+
+ @testing.provide_metadata
+ def _test_get_check_constraints(self, schema=None):
+ orig_meta = self.metadata
+ Table(
+ 'sa_cc', orig_meta,
+ Column('a', Integer()),
+ sa.CheckConstraint('a > 1 AND a < 5', name='cc1'),
+ sa.CheckConstraint('a = 1 OR (a > 2 AND a < 5)', name='cc2'),
+ schema=schema
+ )
+
+ orig_meta.create_all()
+
+ inspector = inspect(orig_meta.bind)
+ reflected = sorted(
+ inspector.get_check_constraints('sa_cc', schema=schema),
+ key=operator.itemgetter('name')
+ )
+
+ reflected = [
+ {"name": item["name"],
+ # trying to minimize effect of quoting, parenthesis, etc.
+ # may need to add more to this as new dialects get CHECK
+ # constraint reflection support
+ "sqltext": re.sub(r"[`'\(\)]", '', item["sqltext"].lower())}
+ for item in reflected
+ ]
+ eq_(
+ reflected,
+ [
+ {'name': 'cc1', 'sqltext': 'a > 1 and a < 5'},
+ {'name': 'cc2', 'sqltext': 'a = 1 or a > 2 and a < 5'}
+ ]
+ )
+
@testing.provide_metadata
def _test_get_view_definition(self, schema=None):
meta = self.metadata

Issue 2:
Migrated issue, originally created by Michael Bayer (@zzzeek)

diff --git a/lib/sqlalchemy/orm/events.py b/lib/sqlalchemy/orm/events.py
index c50a7b0..bd0b48f 100644
--- a/lib/sqlalchemy/orm/events.py
+++ b/lib/sqlalchemy/orm/events.py
@@ -1490,7 +1490,7 @@ class AttributeEvents(event.Events):
@classmethod
def _listen(cls, event_key, active_history=False,
raw=False, retval=False,
- propagate=False):
+ propagate=False, **kw):

target, identifier, fn = \
event_key.dispatch_target, event_key.identifier, event_key.fn
@@ -1509,14 +1509,14 @@ class AttributeEvents(event.Events):
return fn(target, value, *arg)
event_key = event_key.with_wrapper(wrap)

- event_key.base_listen(propagate=propagate)
+ event_key.base_listen(propagate=propagate, **kw)

if propagate:
manager = instrumentation.manager_of_class(target.class_)

for mgr in manager.subclass_managers(True):
event_key.with_dispatch_target(
- mgr[target.key]).base_listen(propagate=True)
+ mgr[target.key]).base_listen(propagate=True, **kw)

def append(self, target, value, initiator):
"""Receive a collection append event.

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

Base = declarative_base()

class A(Base):
__tablename__ = 'a'

id = Column(Integer, primary_key=True)
bs = relationship("B")

class B(Base):
__tablename__ = 'b'

id = Column(Integer, primary_key=True)
a_id = Column(Integer, ForeignKey('a.id'))

e = create_engine("sqlite://", echo=True)

sess = Session(e)

@event.listens_for(A.bs, 'set', named=True)
def update_challenge_points(target, **kw):
pass