- Small adjustment made to the batch handling for reflected CHECK

constraints to accommodate for SQLAlchemy 1.1 now reflecting these.
Batch mode still does not support CHECK constraints from the reflected
table as these can't be easily differentiated from the ones created
by types such as Boolean.

diff --git a/alembic/operations/batch.py b/alembic/operations/batch.py
index 2497e864761739ac4e381f28395c4d9308473710..90fe55776c7bb2c9917b669b1fab7cf212ce5a42 100644 (file)
--- a/alembic/operations/batch.py
+++ b/alembic/operations/batch.py
@@ -1,5 +1,5 @@
-from sqlalchemy import Table, MetaData, Index, select, Column, \
-    ForeignKeyConstraint, cast
+ffrom sqlalchemy import Table, MetaData, Index, select, Column, \
+    ForeignKeyConstraint, cast, CheckConstraint
 from sqlalchemy import types as sqltypes
 from sqlalchemy import schema as sql_schema
 from sqlalchemy.util import OrderedDict
@@ -63,6 +63,7 @@ class BatchOperationsImpl(object):
 
             if self.copy_from is not None:
                 existing_table = self.copy_from
+                reflected = False
             else:
                 existing_table = Table(
                     self.table_name, m1,
@@ -70,9 +71,10 @@ class BatchOperationsImpl(object):
                     autoload=True,
                     autoload_with=self.operations.get_bind(),
                     *self.reflect_args, **self.reflect_kwargs)
+                reflected = True
 
             batch_impl = ApplyBatchImpl(
-                existing_table, self.table_args, self.table_kwargs)
+                existing_table, self.table_args, self.table_kwargs, reflected)
             for opname, arg, kw in self.batch:
                 fn = getattr(batch_impl, opname)
                 fn(*arg, **kw)
@@ -111,7 +113,7 @@ class BatchOperationsImpl(object):
 
 
 class ApplyBatchImpl(object):
-    def __init__(self, table, table_args, table_kwargs):
+    def __init__(self, table, table_args, table_kwargs, reflected):
         self.table = table  # this is a Table object
         self.table_args = table_args
         self.table_kwargs = table_kwargs
@@ -119,6 +121,7 @@ class ApplyBatchImpl(object):
         self.column_transfers = OrderedDict(
             (c.name, {'expr': c}) for c in self.table.c
         )
+        self.reflected = reflected
         self._grab_table_elements()
 
     def _grab_table_elements(self):
@@ -139,6 +142,10 @@ class ApplyBatchImpl(object):
         for const in self.table.constraints:
             if _is_type_bound(const):
                 continue
+            elif self.reflected and isinstance(const, CheckConstraint):
+                # TODO: we are skipping reflected CheckConstraint because
+                # we have no way to determine _is_type_bound() for these.
+                pass
             elif const.name:
                 self.named_constraints[const.name] = const
             else:

diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst
index 097527fd9566c54884f552a1bfdbcbe2daeaebcc..ed4e29d53745197cbc9b9794da36156d19330153 100644 (file)
--- a/docs/build/changelog.rst
+++ b/docs/build/changelog.rst
@@ -3,6 +3,18 @@
 Changelog
 ==========
 
+.. changelog::
+    :version: 0.8.7
+
+    .. change::
+      :tags: bug, batch
+
+      Small adjustment made to the batch handling for reflected CHECK
+      constraints to accommodate for SQLAlchemy 1.1 now reflecting these.
+      Batch mode still does not support CHECK constraints from the reflected
+      table as these can't be easily differentiated from the ones created
+      by types such as Boolean.
+
 .. changelog::
     :version: 0.8.6
     :released: April 14, 2016

diff --git a/tests/test_batch.py b/tests/test_batch.py
index 6e55eaca430025033190921db8c02a5909bf8d5c..490aee370a5398808cf0c4ea0f129b6f1834988d 100644 (file)
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -34,7 +34,7 @@ class BatchApplyTest(TestBase):
             Column('x', String(10)),
             Column('y', Integer)
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _uq_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -45,7 +45,7 @@ class BatchApplyTest(TestBase):
             Column('y', Integer),
             UniqueConstraint('y', name='uq1')
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _ix_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -56,7 +56,7 @@ class BatchApplyTest(TestBase):
             Column('y', Integer),
             Index('ix1', 'y')
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _literal_ck_fixture(
             self, copy_from=None, table_args=(), table_kwargs={}):
@@ -70,7 +70,7 @@ class BatchApplyTest(TestBase):
                 Column('email', String()),
                 CheckConstraint("email LIKE '%@%'")
             )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _sql_ck_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -80,7 +80,7 @@ class BatchApplyTest(TestBase):
             Column('email', String())
         )
         t.append_constraint(CheckConstraint(t.c.email.like('%@%')))
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _fk_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -90,7 +90,7 @@ class BatchApplyTest(TestBase):
             Column('email', String()),
             Column('user_id', Integer, ForeignKey('user.id'))
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _multi_fk_fixture(self, table_args=(), table_kwargs={}, schema=None):
         m = MetaData()
@@ -112,7 +112,7 @@ class BatchApplyTest(TestBase):
                 ['%suser.id' % schemaarg, '%suser.id_version' % schemaarg]),
             schema=schema
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _named_fk_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -122,7 +122,7 @@ class BatchApplyTest(TestBase):
             Column('email', String()),
             Column('user_id', Integer, ForeignKey('user.id', name='ufk'))
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _selfref_fk_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -132,7 +132,7 @@ class BatchApplyTest(TestBase):
             Column('parent_id', Integer, ForeignKey('tname.id')),
             Column('data', String)
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _boolean_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -141,7 +141,7 @@ class BatchApplyTest(TestBase):
             Column('id', Integer, primary_key=True),
             Column('flag', Boolean)
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _boolean_no_ck_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -150,7 +150,7 @@ class BatchApplyTest(TestBase):
             Column('id', Integer, primary_key=True),
             Column('flag', Boolean(create_constraint=False))
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _enum_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -159,7 +159,7 @@ class BatchApplyTest(TestBase):
             Column('id', Integer, primary_key=True),
             Column('thing', Enum('a', 'b', 'c'))
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _server_default_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
@@ -168,7 +168,7 @@ class BatchApplyTest(TestBase):
             Column('id', Integer, primary_key=True),
             Column('thing', String(), server_default='')
         )
-        return ApplyBatchImpl(t, table_args, table_kwargs)
+        return ApplyBatchImpl(t, table_args, table_kwargs, False)
 
     def _assert_impl(self, impl, colnames=None,
                      ddl_contains=None, ddl_not_contains=None,