Support DROP of named check constraint from column for batch

Added support to drop named CHECK constraints that are specified as part of
a column, rather than table wide.  Previously, only constraints associated
with the table were considered.

Change-Id: Id1765357e0fa59745b41ba233b18a53e38358e0b
Fixes: #711

diff --git a/alembic/operations/batch.py b/alembic/operations/batch.py
index c46180167d1cb845551de77dd6a6336ba3d99e37..eb0214c8da6673ebc2a484dcfe5dbe8b748df94c 100644 (file)
--- a/alembic/operations/batch.py
+++ b/alembic/operations/batch.py
@@ -198,8 +198,10 @@ class ApplyBatchImpl(object):
             self.columns[c.name] = c_copy
         self.named_constraints = {}
         self.unnamed_constraints = []
+        self.col_named_constraints = {}
         self.indexes = {}
         self.new_indexes = {}
+
         for const in self.table.constraints:
             if _is_type_bound(const):
                 continue
@@ -212,6 +214,12 @@ class ApplyBatchImpl(object):
             else:
                 self.unnamed_constraints.append(const)
 
+        if not self.reflected:
+            for col in self.table.c:
+                for const in col.constraints:
+                    if const.name:
+                        self.col_named_constraints[const.name] = (col, const)
+
         for idx in self.table.indexes:
             self.indexes[idx.name] = idx
 
@@ -511,7 +519,14 @@ class ApplyBatchImpl(object):
         if not const.name:
             raise ValueError("Constraint must have a name")
         try:
-            const = self.named_constraints.pop(const.name)
+            if const.name in self.col_named_constraints:
+                col, const = self.col_named_constraints.pop(const.name)
+
+                for col_const in list(self.columns[col.name].constraints):
+                    if col_const.name == const.name:
+                        self.columns[col.name].constraints.remove(col_const)
+            else:
+                const = self.named_constraints.pop(const.name)
         except KeyError:
             if _is_type_bound(const):
                 # type-bound constraints are only included in the new

diff --git a/docs/build/unreleased/711.rst b/docs/build/unreleased/711.rst
new file mode 100644 (file)
index 0000000..4117a3a
--- /dev/null
+++ b/docs/build/unreleased/711.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: bug, sqlite, batch
+    :tickets: 711
+
+    Added support to drop named CHECK constraints that are specified as part of
+    a column, rather than table wide.  Previously, only constraints associated
+    with the table were considered.

diff --git a/tests/test_batch.py b/tests/test_batch.py
index d52e0b25a6a8e4043ac2dd98ce7a8788563ef4d2..90d2a4ffd22ed3cc5caa2268cc30c8e55c6f71a6 100644 (file)
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -74,6 +74,29 @@ class BatchApplyTest(TestBase):
         )
         return ApplyBatchImpl(self.impl, t, table_args, table_kwargs, False)
 
+    def _named_ck_table_fixture(self, table_args=(), table_kwargs={}):
+        m = MetaData()
+        t = Table(
+            "tname",
+            m,
+            Column("id", Integer, primary_key=True),
+            Column("x", String()),
+            Column("y", Integer),
+            CheckConstraint("y > 5", name="ck1"),
+        )
+        return ApplyBatchImpl(self.impl, t, table_args, table_kwargs, False)
+
+    def _named_ck_col_fixture(self, table_args=(), table_kwargs={}):
+        m = MetaData()
+        t = Table(
+            "tname",
+            m,
+            Column("id", Integer, primary_key=True),
+            Column("x", String()),
+            Column("y", Integer, CheckConstraint("y > 5", name="ck1")),
+        )
+        return ApplyBatchImpl(self.impl, t, table_args, table_kwargs, False)
+
     def _ix_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
         t = Table(
@@ -731,6 +754,39 @@ class BatchApplyTest(TestBase):
             ddl_not_contains="CONSTRAINT uq1 UNIQUE",
         )
 
+    def test_add_ck(self):
+        impl = self._simple_fixture()
+        ck = self.op.schema_obj.check_constraint("ck1", "tname", "y > 5")
+
+        impl.add_constraint(ck)
+        self._assert_impl(
+            impl,
+            colnames=["id", "x", "y"],
+            ddl_contains="CONSTRAINT ck1 CHECK (y > 5)",
+        )
+
+    def test_drop_ck_table(self):
+        impl = self._named_ck_table_fixture()
+
+        ck = self.op.schema_obj.check_constraint("ck1", "tname", "y > 5")
+        impl.drop_constraint(ck)
+        self._assert_impl(
+            impl,
+            colnames=["id", "x", "y"],
+            ddl_not_contains="CONSTRAINT ck1 CHECK (y > 5)",
+        )
+
+    def test_drop_ck_col(self):
+        impl = self._named_ck_col_fixture()
+
+        ck = self.op.schema_obj.check_constraint("ck1", "tname", "y > 5")
+        impl.drop_constraint(ck)
+        self._assert_impl(
+            impl,
+            colnames=["id", "x", "y"],
+            ddl_not_contains="CONSTRAINT ck1 CHECK (y > 5)",
+        )
+
     def test_create_index(self):
         impl = self._simple_fixture()
         ix = self.op.schema_obj.index("ix1", "tname", ["y"])