- Fixed bug in batch where if the target table contained multiple

foreign keys to the same target table, the batch mechanics would
fail with a "table already exists" error.  Thanks for the help
on this from Lucas Kahlert. fixes #254

3 files changed, 51 insertions, 6 deletions

diff --git a/alembic/batch.py b/alembic/batch.py
index f3ac734..e97f74e 100644
--- a/alembic/batch.py
+++ b/alembic/batch.py
@@ -1,6 +1,7 @@
 from sqlalchemy import Table, MetaData, Index, select, Column, \
     ForeignKeyConstraint, cast
 from sqlalchemy import types as sqltypes
+from sqlalchemy import schema as sql_schema
 from sqlalchemy.util import OrderedDict
 from . import util
 from .ddl.base import _columns_for_constraint, _is_type_bound
@@ -174,12 +175,22 @@ class ApplyBatchImpl(object):
         else:
             referent_schema = None
         if tname != '_alembic_batch_temp':
-            Table(
-                tname, metadata,
-                *[Column(n, sqltypes.NULLTYPE) for n in
-                    [elem._get_colspec().split(".")[-1]
-                     for elem in constraint.elements]],
-                schema=referent_schema)
+            key = sql_schema._get_table_key(tname, referent_schema)
+            if key in metadata.tables:
+                t = metadata.tables[key]
+                for elem in constraint.elements:
+                    colname = elem._get_colspec().split(".")[-1]
+                    if not t.c.contains_column(colname):
+                        t.append_column(
+                            Column(colname, sqltypes.NULLTYPE)
+                        )
+            else:
+                Table(
+                    tname, metadata,
+                    *[Column(n, sqltypes.NULLTYPE) for n in
+                        [elem._get_colspec().split(".")[-1]
+                         for elem in constraint.elements]],
+                    schema=referent_schema)
 
     def _create(self, op_impl):
         self._transfer_elements_to_new_table()
diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst
index 145cab6..44e2e07 100644
--- a/docs/build/changelog.rst
+++ b/docs/build/changelog.rst
@@ -7,6 +7,15 @@ Changelog
     :version: 0.7.2
 
     .. change::
+      :tags: bug, batch
+      :tickets: 254
+
+      Fixed bug in batch where if the target table contained multiple
+      foreign keys to the same target table, the batch mechanics would
+      fail with a "table already exists" error.  Thanks for the help
+      on this from Lucas Kahlert.
+
+    .. change::
       :tags: bug, mysql
       :tickets: 251
       :pullreq: bitbucket:35
diff --git a/tests/test_batch.py b/tests/test_batch.py
index 70dad56..279f945 100644
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -89,6 +89,22 @@ class BatchApplyTest(TestBase):
         )
         return ApplyBatchImpl(t, table_args, table_kwargs)
 
+    def _multi_fk_fixture(self, table_args=(), table_kwargs={}):
+        m = MetaData()
+        t = Table(
+            'tname', m,
+            Column('id', Integer, primary_key=True),
+            Column('email', String()),
+            Column('user_id_1', Integer, ForeignKey('user.id')),
+            Column('user_id_2', Integer, ForeignKey('user.id')),
+            Column('user_id_3', Integer),
+            Column('user_id_version', Integer),
+            ForeignKeyConstraint(
+                ['user_id_3', 'user_id_version'],
+                ['user.id', 'user.id_version'])
+        )
+        return ApplyBatchImpl(t, table_args, table_kwargs)
+
     def _named_fk_fixture(self, table_args=(), table_kwargs={}):
         m = MetaData()
         t = Table(
@@ -321,6 +337,15 @@ class BatchApplyTest(TestBase):
             "user.id"
         )
 
+    def test_regen_multi_fk(self):
+        impl = self._multi_fk_fixture()
+        self._assert_impl(
+            impl, colnames=[
+                'id', 'email', 'user_id_1', 'user_id_2',
+                'user_id_3', 'user_id_version'],
+            ddl_contains='FOREIGN KEY(user_id_3, user_id_version) '
+            'REFERENCES "user" (id, id_version)')
+
     def test_drop_col(self):
         impl = self._simple_fixture()
         impl.drop_column('tname', column('x'))