Commits

Mike Bayer committed 0de174c

- [bug] Added "type" argument to op.drop_constraint(),
and implemented full constraint drop support for
MySQL. CHECK and undefined raise an error.
MySQL needs the constraint type
in order to emit a DROP CONSTRAINT. #44

  • Participants
  • Parent commits 05f0fae

Comments (0)

Files changed (5)

+0.3.3
+=====
+- [bug] Added "type" argument to op.drop_constraint(),
+  and implemented full constraint drop support for
+  MySQL.  CHECK and undefined raise an error.  
+  MySQL needs the constraint type
+  in order to emit a DROP CONSTRAINT. #44
+
 0.3.2
 =====
 - [feature] Basic support for Oracle added, 

alembic/__init__.py

 from os import path
 
-__version__ = '0.3.2'
+__version__ = '0.3.3'
 
 package_dir = path.abspath(path.dirname(__file__))
 

alembic/ddl/mysql.py

 from alembic.ddl.base import alter_table
 from alembic import util
 from sqlalchemy import types as sqltypes
+from sqlalchemy import schema
 
 class MySQLImpl(DefaultImpl):
     __dialect__ = 'mysql'
 
     return spec
 
+@compiles(schema.DropConstraint, "mysql")
+def _mysql_drop_constraint(element, compiler, **kw):
+    """Redefine SQLAlchemy's drop constraint to 
+    raise errors for invalid constraint type."""
 
+    constraint = element.element
+    if isinstance(constraint, (schema.ForeignKeyConstraint,
+                                schema.PrimaryKeyConstraint,
+                                schema.UniqueConstraint)
+                                ):
+        return compiler.visit_drop_constraint(element, **kw)
+    elif isinstance(constraint, schema.CheckConstraint):
+        raise NotImplementedError(
+                "MySQL does not support CHECK constraints.")
+    else:
+        raise NotImplementedError(
+                "No generic 'DROP CONSTRAINT' in MySQL - "
+                "please specify constraint type")
+

alembic/operations.py

         # 0.7.6 and further raises on Index with no columns
         self.impl.drop_index(self._index(name, tablename, ['x']))
 
-    def drop_constraint(self, name, tablename):
-        """Drop a constraint of the given name"""
+    def drop_constraint(self, name, tablename, type=None):
+        """Drop a constraint of the given name, typically via DROP CONSTRAINT.
+        
+        :param name: name of the constraint.
+        :param tablename: tablename.
+        :param type: optional, required on MySQL.  can be 
+        'foreignkey', 'unique', or 'check'
+
+        """
         t = self._table(tablename)
-        const = schema.Constraint(name=name)
+        types = {
+            'foreignkey':lambda name:schema.ForeignKeyConstraint(
+                                [], [], name=name),
+            'unique':schema.UniqueConstraint,
+            'check':lambda name:schema.CheckConstraint("", name=name),
+            None:schema.Constraint
+        }
+        try:
+            const = types[type]
+        except KeyError:
+            raise TypeError("'type' can be one of %s" % 
+                        ", ".join(sorted(repr(x) for x in types))) 
+
+        const = const(name=name)
         t.append_constraint(const)
         self.impl.drop_constraint(const)
 

tests/test_mysql.py

         "All MySQL ALTER COLUMN operations require the existing type.",
         op.alter_column, 't1', 'c1', nullable=False, server_default="q"
     )
+
+def test_drop_fk():
+    context = op_fixture('mysql')
+    op.drop_constraint("f1", "t1", "foreignkey")
+    context.assert_(
+        "ALTER TABLE t1 DROP FOREIGN KEY f1"
+    )
+
+def test_drop_unique():
+    context = op_fixture('mysql')
+    op.drop_constraint("f1", "t1", "unique")
+    context.assert_(
+        "ALTER TABLE t1 DROP INDEX f1"
+    )
+
+def test_drop_check():
+    context = op_fixture('mysql')
+    assert_raises_message(
+        NotImplementedError,
+        "MySQL does not support CHECK constraints.",
+        op.drop_constraint, "f1", "t1", "check"
+    )
+
+def test_drop_unknown():
+    context = op_fixture('mysql')
+    assert_raises_message(
+        TypeError,
+        "'type' can be one of 'check', 'foreignkey', 'unique', None",
+        op.drop_constraint, "f1", "t1", "typo"
+    )
+
+def test_drop_generic_constraint():
+    context = op_fixture('mysql')
+    assert_raises_message(
+        NotImplementedError,
+        "No generic 'DROP CONSTRAINT' in MySQL - please "
+        "specify constraint type",
+        op.drop_constraint, "f1", "t1"
+    )