From 93efb62fd100f5135928443c2c325ae78b1c1fd0 Mon Sep 17 00:00:00 2001
From: Matt Riedemann <mriedem@us.ibm.com>
Date: Mon, 14 Apr 2014 21:20:05 -0700
Subject: [PATCH] Move patch from oslo to drop unique constraints with sqlite

oslo-incubator commit 3f503faac for making sqlite work with dropping
unique constraints in database migrations. This was made in
oslo-incubator since at the time sqlalchemy-migrate was not in
stackforge. Now that we can update sqlalchemy-migrate, move the patch
over from oslo.

This change also adds the support for the case that a unique constraint
is dropped because the column it's on is dropped.

Note that there are already unit tests that cover dropping a unique
constraint directly and implicitly via dropping a column that is in
the unique constraint.

Related-Bug: #1307266

Change-Id: I5ee8082a83aebf66f6e1dacb093ed79e13f73f5e
---
 migrate/changeset/databases/sqlite.py      | 51 ++++++++++++++++++++--
 migrate/tests/changeset/test_constraint.py |  2 +-
 2 files changed, 49 insertions(+), 4 deletions(-)

diff --git a/migrate/changeset/databases/sqlite.py b/migrate/changeset/databases/sqlite.py
index 6453422..2739f9f 100644
--- a/migrate/changeset/databases/sqlite.py
+++ b/migrate/changeset/databases/sqlite.py
@@ -5,8 +5,10 @@
 """
 from UserDict import DictMixin
 from copy import copy
+import re
 
 from sqlalchemy.databases import sqlite as sa_base
+from sqlalchemy.schema import UniqueConstraint
 
 from migrate import exceptions
 from migrate.changeset import ansisql
@@ -24,7 +26,38 @@ class SQLiteCommon(object):
 
 class SQLiteHelper(SQLiteCommon):
 
-    def recreate_table(self,table,column=None,delta=None):
+    def _get_unique_constraints(self, table):
+        """Retrieve information about existing unique constraints of the table
+
+        This feature is needed for recreate_table() to work properly.
+        """
+
+        data = table.metadata.bind.execute(
+            """SELECT sql
+               FROM sqlite_master
+               WHERE
+                   type='table' AND
+                   name=:table_name""",
+            table_name=table.name
+        ).fetchone()[0]
+
+        UNIQUE_PATTERN = "CONSTRAINT (\w+) UNIQUE \(([^\)]+)\)"
+        constraints = []
+        for name, cols in re.findall(UNIQUE_PATTERN, data):
+            # Filter out any columns that were dropped from the table.
+            columns = []
+            for c in cols.split(","):
+                if c in table.columns:
+                    # There was a bug in reflection of SQLite columns with
+                    # reserved identifiers as names (SQLite can return them
+                    # wrapped with double quotes), so strip double quotes.
+                    columns.extend(c.strip(' "'))
+            if columns:
+                constraints.extend(UniqueConstraint(*columns, name=name))
+        return constraints
+
+    def recreate_table(self, table, column=None, delta=None,
+                       omit_uniques=None):
         table_name = self.preparer.format_table(table)
 
         # we remove all indexes so as not to have
@@ -32,6 +65,15 @@ class SQLiteHelper(SQLiteCommon):
         for index in table.indexes:
             index.drop()
 
+        # reflect existing unique constraints
+        for uc in self._get_unique_constraints(table):
+            table.append_constraint(uc)
+        # omit given unique constraints when creating a new table if required
+        table.constraints = set([
+            cons for cons in table.constraints
+            if omit_uniques is None or cons.name not in omit_uniques
+        ])
+
         self.append('ALTER TABLE %s RENAME TO migration_tmp' % table_name)
         self.execute()
 
@@ -123,9 +165,12 @@ class SQLiteConstraintGenerator(ansisql.ANSIConstraintGenerator, SQLiteHelper, S
 
 
 class SQLiteConstraintDropper(ansisql.ANSIColumnDropper,
-                              SQLiteCommon,
+                              SQLiteHelper,
                               ansisql.ANSIConstraintCommon):
 
+    def _modify_table(self, table, column, delta):
+        return 'INSERT INTO %(table_name)s SELECT * from migration_tmp'
+
     def visit_migrate_primary_key_constraint(self, constraint):
         tmpl = "DROP INDEX %s "
         name = self.get_constraint_name(constraint)
@@ -140,7 +185,7 @@ class SQLiteConstraintDropper(ansisql.ANSIColumnDropper,
         self._not_supported('ALTER TABLE DROP CONSTRAINT')
 
     def visit_migrate_unique_constraint(self, *p, **k):
-        self._not_supported('ALTER TABLE DROP CONSTRAINT')
+        self.recreate_table(p[0].table, omit_uniques=[p[0].name])
 
 
 # TODO: technically primary key is a NOT NULL + UNIQUE constraint, should add NOT NULL to index
diff --git a/migrate/tests/changeset/test_constraint.py b/migrate/tests/changeset/test_constraint.py
index d27554e..6421206 100644
--- a/migrate/tests/changeset/test_constraint.py
+++ b/migrate/tests/changeset/test_constraint.py
@@ -274,7 +274,7 @@ class TestAutoname(CommonTestConstraint):
         self.table.insert(values={'id': 2, 'fkey': 2}).execute()
         self.table.insert(values={'id': 1, 'fkey': 3}).execute()
 
-    @fixture.usedb(not_supported=['oracle', 'sqlite'])
+    @fixture.usedb(not_supported=['oracle'])
     def test_autoname_unique(self):
         """UniqueConstraints can guess their name if None is given"""
         cons = UniqueConstraint(self.table.c.fkey)