- Fully implemented the

:paramref:`~.Operations.batch_alter_table.copy_from` parameter for
batch mode, which previously was not functioning.  This allows
"batch mode" to be usable in conjunction with ``--sql``.
fixes #289
- sqlite dialect checks for "create_index" and "drop_index" as exceptions
for "recreate" in batch mode, the same way as "add_column", so that
unnecessary table recreates don't emit for index-only operations

7 files changed, 182 insertions, 11 deletions

diff --git a/alembic/batch.py b/alembic/batch.py
index 6e5dc75..1006739 100644
--- a/alembic/batch.py
+++ b/alembic/batch.py
@@ -58,12 +58,15 @@ class BatchOperationsImpl(object):
             else:
                 m1 = MetaData()
 
-            existing_table = Table(
-                self.table_name, m1,
-                schema=self.schema,
-                autoload=True,
-                autoload_with=self.operations.get_bind(),
-                *self.reflect_args, **self.reflect_kwargs)
+            if self.copy_from is not None:
+                existing_table = self.copy_from
+            else:
+                existing_table = Table(
+                    self.table_name, m1,
+                    schema=self.schema,
+                    autoload=True,
+                    autoload_with=self.operations.get_bind(),
+                    *self.reflect_args, **self.reflect_kwargs)
 
             batch_impl = ApplyBatchImpl(
                 existing_table, self.table_args, self.table_kwargs)
diff --git a/alembic/ddl/sqlite.py b/alembic/ddl/sqlite.py
index 16beddf..5d231b5 100644
--- a/alembic/ddl/sqlite.py
+++ b/alembic/ddl/sqlite.py
@@ -21,7 +21,7 @@ class SQLiteImpl(DefaultImpl):
 
         """
         for op in batch_op.batch:
-            if op[0] != 'add_column':
+            if op[0] not in ('add_column', 'create_index', 'drop_index'):
                 return True
         else:
             return False
diff --git a/alembic/operations.py b/alembic/operations.py
index 683d2bd..485943e 100644
--- a/alembic/operations.py
+++ b/alembic/operations.py
@@ -242,20 +242,28 @@ class Operations(object):
 
         .. note::  The table copy operation will currently not copy
            CHECK constraints, and may not copy UNIQUE constraints that are
-           unnamed, as is possible on SQLite.
+           unnamed, as is possible on SQLite.   See the section
+           :ref:`sqlite_batch_constraints` for workarounds.
 
         :param table_name: name of table
         :param schema: optional schema name.
         :param recreate: under what circumstances the table should be
          recreated. At its default of ``"auto"``, the SQLite dialect will
-         recreate the table if any operations other than ``add_column()`` are
+         recreate the table if any operations other than ``add_column()``,
+         ``create_index()``, or ``drop_index()`` are
          present. Other options include ``"always"`` and ``"never"``.
         :param copy_from: optional :class:`~sqlalchemy.schema.Table` object
          that will act as the structure of the table being copied.  If omitted,
          table reflection is used to retrieve the structure of the table.
 
+         .. versionadded:: 0.7.6 Fully implemented the
+            :paramref:`~.Operations.batch_alter_table.copy_from`
+            parameter.
+
          .. seealso::
 
+            :ref:`batch_offline_mode`
+
             :paramref:`~.Operations.batch_alter_table.reflect_args`
 
             :paramref:`~.Operations.batch_alter_table.reflect_kwargs`
diff --git a/alembic/testing/fixtures.py b/alembic/testing/fixtures.py
index 6336967..4091388 100644
--- a/alembic/testing/fixtures.py
+++ b/alembic/testing/fixtures.py
@@ -100,7 +100,17 @@ def op_fixture(dialect='default', as_sql=False, naming_convention=None):
             # TODO: this might need to
             # be more like a real connection
             # as tests get more involved
-            self.connection = mock.Mock(dialect=dialect)
+            if as_sql and self.dialect.name != 'default':
+                # act similarly to MigrationContext
+                def dump(construct, *multiparams, **params):
+                    self._exec(construct)
+
+                self.connection = create_engine(
+                    "%s://" % self.dialect.name,
+                    strategy="mock", executor=dump)
+
+            else:
+                self.connection = mock.Mock(dialect=dialect)
 
         def _exec(self, construct, *args, **kw):
             if isinstance(construct, string_types):
@@ -128,6 +138,9 @@ def op_fixture(dialect='default', as_sql=False, naming_convention=None):
             self.opts = opts
             self.as_sql = as_sql
 
+        def clear_assertions(self):
+            self.impl.assertion[:] = []
+
         def assert_(self, *sql):
             # TODO: make this more flexible about
             # whitespace and such
diff --git a/docs/build/batch.rst b/docs/build/batch.rst
index 307d2a1..64eeefb 100644
--- a/docs/build/batch.rst
+++ b/docs/build/batch.rst
@@ -110,6 +110,8 @@ pre-fabricated :class:`~sqlalchemy.schema.Table` object; see
    added :paramref:`.Operations.batch_alter_table.reflect_args`
    and :paramref:`.Operations.batch_alter_table.reflect_kwargs` options.
 
+.. _sqlite_batch_constraints:
+
 Dealing with Constraints
 ------------------------
 
@@ -251,6 +253,10 @@ preferred style of working; however, if one needs to do SQLite-compatible
 "move and copy" migrations and need them to generate flat SQL files in
 "offline" mode, there's not much alternative.
 
+.. versionadded:: 0.7.6 Fully implemented the
+   :paramref:`~.Operations.batch_alter_table.copy_from`
+   parameter.
+
 
 Batch mode with Autogenerate
 ----------------------------
diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst
index 1d730e1..9b27cc7 100644
--- a/docs/build/changelog.rst
+++ b/docs/build/changelog.rst
@@ -8,11 +8,26 @@ Changelog
 
     .. change::
       :tags: bug, batch
+      :tickets: 289
+
+      Fully implemented the
+      :paramref:`~.Operations.batch_alter_table.copy_from` parameter for
+      batch mode, which previously was not functioning.  This allows
+      "batch mode" to be usable in conjunction with ``--sql``.
+
+    .. change::
+      :tags: bug, batch
       :tickets: 287
 
       Repaired support for the :meth:`.BatchOperations.create_index`
       directive, which was mis-named internally such that the operation
-      within a batch context could not proceed.
+      within a batch context could not proceed.   The create index
+      operation will proceed as part of a larger "batch table recreate"
+      operation only if
+      :paramref:`~.Operations.batch_alter_table.recreate` is set to
+      "always", or if the batch operation includes other instructions that
+      require a table recreate.
+
 
 .. changelog::
     :version: 0.7.5
diff --git a/tests/test_batch.py b/tests/test_batch.py
index 76f0c12..ffd88cb 100644
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -1,6 +1,8 @@
 from contextlib import contextmanager
 import re
 
+import io
+
 from alembic.testing import exclusions
 from alembic.testing import TestBase, eq_, config
 from alembic.testing.fixtures import op_fixture
@@ -9,6 +11,7 @@ from alembic.operations import Operations
 from alembic.batch import ApplyBatchImpl
 from alembic.migration import MigrationContext
 
+
 from sqlalchemy import inspect
 from sqlalchemy import Integer, Table, Column, String, MetaData, ForeignKey, \
     UniqueConstraint, ForeignKeyConstraint, Index, Boolean, CheckConstraint, \
@@ -641,6 +644,129 @@ class BatchAPITest(TestBase):
         )
 
 
+class CopyFromTest(TestBase):
+    __requires__ = ('sqlalchemy_08', )
+
+    def _fixture(self):
+        self.metadata = MetaData()
+        self.table = Table(
+            'foo', self.metadata,
+            Column('id', Integer, primary_key=True),
+            Column('data', String(50)),
+            Column('x', Integer),
+        )
+
+        context = op_fixture(dialect="sqlite", as_sql=True)
+        self.op = Operations(context)
+        return context
+
+    def test_change_type(self):
+        context = self._fixture()
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table) as batch_op:
+            batch_op.alter_column('data', type_=Integer)
+
+        context.assert_(
+            'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+            'data INTEGER, x INTEGER, PRIMARY KEY (id))',
+            'INSERT INTO _alembic_batch_temp (id, data, x) SELECT foo.id, '
+            'CAST(foo.data AS INTEGER) AS anon_1, foo.x FROM foo',
+            'DROP TABLE foo',
+            'ALTER TABLE _alembic_batch_temp RENAME TO foo'
+        )
+
+    def test_create_drop_index_w_always(self):
+        context = self._fixture()
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table, recreate='always') as batch_op:
+            batch_op.create_index(
+                batch_op.f('ix_data'), ['data'], unique=True)
+
+        context.assert_(
+            'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+            'data VARCHAR(50), '
+            'x INTEGER, PRIMARY KEY (id))',
+            'CREATE UNIQUE INDEX ix_data ON _alembic_batch_temp (data)',
+            'INSERT INTO _alembic_batch_temp (id, data, x) '
+            'SELECT foo.id, foo.data, foo.x FROM foo',
+            'DROP TABLE foo',
+            'ALTER TABLE _alembic_batch_temp RENAME TO foo'
+        )
+
+        context.clear_assertions()
+
+        Index('ix_data', self.table.c.data, unique=True)
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table, recreate='always') as batch_op:
+            batch_op.drop_index('ix_data')
+
+        context.assert_(
+            'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+            'data VARCHAR(50), x INTEGER, PRIMARY KEY (id))',
+            'INSERT INTO _alembic_batch_temp (id, data, x) '
+            'SELECT foo.id, foo.data, foo.x FROM foo',
+            'DROP TABLE foo',
+            'ALTER TABLE _alembic_batch_temp RENAME TO foo'
+        )
+
+    def test_create_drop_index_wo_always(self):
+        context = self._fixture()
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table) as batch_op:
+            batch_op.create_index(
+                batch_op.f('ix_data'), ['data'], unique=True)
+
+        context.assert_(
+            'CREATE UNIQUE INDEX ix_data ON foo (data)'
+        )
+
+        context.clear_assertions()
+
+        Index('ix_data', self.table.c.data, unique=True)
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table) as batch_op:
+            batch_op.drop_index('ix_data')
+
+        context.assert_(
+            'DROP INDEX ix_data'
+        )
+
+    def test_create_drop_index_w_other_ops(self):
+        context = self._fixture()
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table) as batch_op:
+            batch_op.alter_column('data', type_=Integer)
+            batch_op.create_index(
+                batch_op.f('ix_data'), ['data'], unique=True)
+
+        context.assert_(
+            'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+            'data INTEGER, x INTEGER, PRIMARY KEY (id))',
+            'CREATE UNIQUE INDEX ix_data ON _alembic_batch_temp (data)',
+            'INSERT INTO _alembic_batch_temp (id, data, x) SELECT foo.id, '
+            'CAST(foo.data AS INTEGER) AS anon_1, foo.x FROM foo',
+            'DROP TABLE foo',
+            'ALTER TABLE _alembic_batch_temp RENAME TO foo'
+        )
+
+        context.clear_assertions()
+
+        Index('ix_data', self.table.c.data, unique=True)
+        with self.op.batch_alter_table(
+                "foo", copy_from=self.table) as batch_op:
+            batch_op.drop_index('ix_data')
+            batch_op.alter_column('data', type_=String)
+
+        context.assert_(
+            'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+            'data VARCHAR, x INTEGER, PRIMARY KEY (id))',
+            'INSERT INTO _alembic_batch_temp (id, data, x) SELECT foo.id, '
+            'CAST(foo.data AS VARCHAR) AS anon_1, foo.x FROM foo',
+            'DROP TABLE foo',
+            'ALTER TABLE _alembic_batch_temp RENAME TO foo'
+        )
+
+
 class BatchRoundTripTest(TestBase):
     __requires__ = ('sqlalchemy_08', )
     __only_on__ = "sqlite"