from sqlalchemy import types as sqltypes
from sqlalchemy.util import OrderedDict
from . import util
+from .ddl.base import _columns_for_constraint, _is_type_bound
class BatchOperationsImpl(object):
def __init__(self, operations, table_name, schema, recreate,
- copy_from, table_args, table_kwargs):
+ copy_from, table_args, table_kwargs,
+ reflect_args, reflect_kwargs):
if not util.sqla_08:
raise NotImplementedError(
"batch mode requires SQLAlchemy 0.8 or greater.")
self.copy_from = copy_from
self.table_args = table_args
self.table_kwargs = table_kwargs
+ self.reflect_args = reflect_args
+ self.reflect_kwargs = reflect_kwargs
self.batch = []
@property
fn(*arg, **kw)
else:
m1 = MetaData()
+
existing_table = Table(
- self.table_name, m1, schema=self.schema,
- autoload=True, autoload_with=self.operations.get_bind())
+ 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)
self.unnamed_constraints = []
self.indexes = {}
for const in self.table.constraints:
+ if _is_type_bound(const):
+ continue
if const.name:
self.named_constraints[const.name] = const
else:
self.unnamed_constraints:
const_columns = set([
- c.key for c in self._constraint_columns(const)])
+ c.key for c in _columns_for_constraint(const)])
if not const_columns.issubset(self.column_transfers):
continue
*[new_table.c[col] for col in index.columns.keys()],
**index.kwargs)
- def _constraint_columns(self, constraint):
- if isinstance(constraint, ForeignKeyConstraint):
- return [fk.parent for fk in constraint.elements]
- else:
- return list(constraint.columns)
-
def _setup_referent(self, metadata, constraint):
spec = constraint.elements[0]._get_colspec()
parts = spec.split(".")
@contextmanager
def batch_alter_table(
self, table_name, schema=None, recreate="auto", copy_from=None,
- table_args=(), table_kwargs=util.immutabledict()):
+ table_args=(), table_kwargs=util.immutabledict(),
+ reflect_args=(), reflect_kwargs=util.immutabledict()):
"""Invoke a series of per-table migrations in batch.
Batch mode allows a series of operations specific to a table
: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.
+
+ .. seealso::
+
+ :paramref:`~.Operations.batch_alter_table.reflect_args`
+
+ :paramref:`~.Operations.batch_alter_table.reflect_kwargs`
+
+ :param reflect_args: a sequence of additional positional arguments that
+ will be applied to the table structure being reflected / copied;
+ this may be used to pass column and constraint overrides to the
+ table that will be reflected, in lieu of passing the whole
+ :class:`~sqlalchemy.schema.Table` using
+ :paramref:`~.Operations.batch_alter_table.copy_from`.
+
+ .. versionadded:: 0.7.1
+
+ :param reflect_kwargs: a dictionary of additional keyword arguments
+ that will be applied to the table structure being copied; this may be
+ used to pass additional table and reflection options to the table that
+ will be reflected, in lieu of passing the whole
+ :class:`~sqlalchemy.schema.Table` using
+ :paramref:`~.Operations.batch_alter_table.copy_from`.
+
+ .. versionadded:: 0.7.1
+
:param table_args: a sequence of additional positional arguments that
will be applied to the new :class:`~sqlalchemy.schema.Table` when
created, in addition to those copied from the source table.
"""
impl = batch.BatchOperationsImpl(
self, table_name, schema, recreate,
- copy_from, table_args, table_kwargs)
+ copy_from, table_args, table_kwargs, reflect_args, reflect_kwargs)
batch_op = BatchOperations(self.migration_context, impl=impl)
yield batch_op
impl.flush()
to run "move and copy" unconditionally in all cases, including on databases
other than SQLite; more on this is below.
+.. _batch_controlling_table_reflection:
+
+Controlling Table Reflection
+----------------------------
+
+The :class:`~sqlalchemy.schema.Table` object that is reflected when
+"move and copy" proceeds is performed using the standard ``autoload=True``
+approach. This call can be affected using the
+:paramref:`~.Operations.batch_alter_table.reflect_args` and
+:paramref:`~.Operations.batch_alter_table.reflect_kwargs` arguments.
+For example, to override a :class:`~sqlalchemy.schema.Column` within
+the reflection process such that a :class:`~sqlalchemy.types.Boolean`
+object is reflected with the ``create_constraint`` flag set to ``False``::
+
+ with self.op.batch_alter_table(
+ "bar",
+ reflect_args=[Column('flag', Boolean(create_constraint=False))]
+ ) as batch_op:
+ batch_op.alter_column(
+ 'flag', new_column_name='bflag', existing_type=Boolean)
+
+Another use case, add a listener to the :class:`~sqlalchemy.schema.Table`
+as it is reflected so that special logic can be applied to columns or
+types, using the :meth:`~sqlalchemy.events.DDLEvents.column_reflect` event::
+
+ def listen_for_reflect(inspector, table, column_info):
+ "correct an ENUM type"
+ if column_info['name'] == 'my_enum':
+ column_info['type'] = Enum('a', 'b', 'c')
+
+ with self.op.batch_alter_table(
+ "bar",
+ reflect_kwargs=dict(
+ listeners=[
+ ('column_reflect', listen_for_reflect)
+ ]
+ )
+ ) as batch_op:
+ batch_op.alter_column(
+ 'flag', new_column_name='bflag', existing_type=Boolean)
+
+The reflection process may also be bypassed entirely by sending a
+pre-fabricated :class:`~sqlalchemy.schema.Table` object; see
+:ref:`batch_offline_mode` for an example.
+
+.. versionadded:: 0.7.1
+ added :paramref:`.Operations.batch_alter_table.reflect_args`
+ and :paramref:`.Operations.batch_alter_table.reflect_kwargs` options.
+
Dealing with Constraints
------------------------
in the first place, or again specified within
:paramref:`.Operations.batch_alter_table.table_args`.
+.. _batch_offline_mode:
+
Working in Offline Mode
-----------------------
To support offline mode, the system must work without table reflection
present, which means the full table as it intends to be created must be
passed to :meth:`.Operations.batch_alter_table` using
-:paramref:`.Operations.batch_alter_table.copy_from`::
+:paramref:`~.Operations.batch_alter_table.copy_from`::
meta = MetaData()
some_table = Table(
from alembic.migration import MigrationContext
from sqlalchemy import Integer, Table, Column, String, MetaData, ForeignKey, \
- UniqueConstraint, ForeignKeyConstraint, Index
+ UniqueConstraint, ForeignKeyConstraint, Index, Boolean, CheckConstraint, \
+ Enum
from sqlalchemy.sql import column
from sqlalchemy.schema import CreateTable, CreateIndex
)
return ApplyBatchImpl(t, table_args, table_kwargs)
+ def _literal_ck_fixture(
+ self, copy_from=None, table_args=(), table_kwargs={}):
+ m = MetaData()
+ if copy_from is not None:
+ t = copy_from
+ else:
+ t = Table(
+ 'tname', m,
+ Column('id', Integer, primary_key=True),
+ Column('email', String()),
+ CheckConstraint("email LIKE '%@%'")
+ )
+ return ApplyBatchImpl(t, table_args, table_kwargs)
+
+ def _sql_ck_fixture(self, table_args=(), table_kwargs={}):
+ m = MetaData()
+ t = Table(
+ 'tname', m,
+ Column('id', Integer, primary_key=True),
+ Column('email', String())
+ )
+ t.append_constraint(CheckConstraint(t.c.email.like('%@%')))
+ return ApplyBatchImpl(t, table_args, table_kwargs)
+
def _fk_fixture(self, table_args=(), table_kwargs={}):
m = MetaData()
t = Table(
)
return ApplyBatchImpl(t, table_args, table_kwargs)
+ def _boolean_fixture(self, table_args=(), table_kwargs={}):
+ m = MetaData()
+ t = Table(
+ 'tname', m,
+ Column('id', Integer, primary_key=True),
+ Column('flag', Boolean)
+ )
+ return ApplyBatchImpl(t, table_args, table_kwargs)
+
+ def _boolean_no_ck_fixture(self, table_args=(), table_kwargs={}):
+ m = MetaData()
+ t = Table(
+ 'tname', m,
+ Column('id', Integer, primary_key=True),
+ Column('flag', Boolean(create_constraint=False))
+ )
+ return ApplyBatchImpl(t, table_args, table_kwargs)
+
+ def _enum_fixture(self, table_args=(), table_kwargs={}):
+ m = MetaData()
+ t = Table(
+ 'tname', m,
+ Column('id', Integer, primary_key=True),
+ Column('thing', Enum('a', 'b', 'c'))
+ )
+ return ApplyBatchImpl(t, table_args, table_kwargs)
+
def _assert_impl(self, impl, colnames=None,
ddl_contains=None, ddl_not_contains=None,
dialect='default'):
"CAST(tname.%s AS %s) AS anon_1" % (
name, impl.new_table.c[name].type)
if (
- impl.new_table.c[name].type
- is not impl.table.c[name].type)
+ impl.new_table.c[name].type._type_affinity
+ is not impl.table.c[name].type._type_affinity)
else "tname.%s" % name
for name in colnames if name in impl.table.c
)
new_table = self._assert_impl(impl)
eq_(new_table.c.x.name, 'q')
+ def test_rename_col_boolean(self):
+ impl = self._boolean_fixture()
+ impl.alter_column('tname', 'flag', name='bflag')
+ new_table = self._assert_impl(
+ impl, ddl_contains="CHECK (bflag IN (0, 1)",
+ colnames=["id", "flag"])
+ eq_(new_table.c.flag.name, 'bflag')
+ eq_(
+ len([
+ const for const
+ in new_table.constraints
+ if isinstance(const, CheckConstraint)]),
+ 1)
+
+ def test_rename_col_boolean_no_ck(self):
+ impl = self._boolean_no_ck_fixture()
+ impl.alter_column('tname', 'flag', name='bflag')
+ new_table = self._assert_impl(
+ impl, ddl_not_contains="CHECK",
+ colnames=["id", "flag"])
+ eq_(new_table.c.flag.name, 'bflag')
+ eq_(
+ len([
+ const for const
+ in new_table.constraints
+ if isinstance(const, CheckConstraint)]),
+ 0)
+
+ def test_rename_col_enum(self):
+ impl = self._enum_fixture()
+ impl.alter_column('tname', 'thing', name='thang')
+ new_table = self._assert_impl(
+ impl, ddl_contains="CHECK (thang IN ('a', 'b', 'c')",
+ colnames=["id", "thing"])
+ eq_(new_table.c.thing.name, 'thang')
+ eq_(
+ len([
+ const for const
+ in new_table.constraints
+ if isinstance(const, CheckConstraint)]),
+ 1)
+
+ def test_rename_col_literal_ck(self):
+ impl = self._literal_ck_fixture()
+ impl.alter_column('tname', 'email', name='emol')
+ new_table = self._assert_impl(
+ # note this is wrong, we don't dig into the SQL
+ impl, ddl_contains="CHECK (email LIKE '%@%')",
+ colnames=["id", "email"])
+ eq_(
+ len([c for c in new_table.constraints
+ if isinstance(c, CheckConstraint)]), 1)
+
+ eq_(new_table.c.email.name, 'emol')
+
+ def test_rename_col_literal_ck_workaround(self):
+ impl = self._literal_ck_fixture(
+ copy_from=Table(
+ 'tname', MetaData(),
+ Column('id', Integer, primary_key=True),
+ Column('email', String),
+ ),
+ table_args=[CheckConstraint("emol LIKE '%@%'")])
+
+ impl.alter_column('tname', 'email', name='emol')
+ new_table = self._assert_impl(
+ impl, ddl_contains="CHECK (emol LIKE '%@%')",
+ colnames=["id", "email"])
+ eq_(
+ len([c for c in new_table.constraints
+ if isinstance(c, CheckConstraint)]), 1)
+ eq_(new_table.c.email.name, 'emol')
+
+ def test_rename_col_sql_ck(self):
+ impl = self._sql_ck_fixture()
+
+ impl.alter_column('tname', 'email', name='emol')
+ new_table = self._assert_impl(
+ impl, ddl_contains="CHECK (emol LIKE '%@%')",
+ colnames=["id", "email"])
+ eq_(
+ len([c for c in new_table.constraints
+ if isinstance(c, CheckConstraint)]), 1)
+
+ eq_(new_table.c.email.name, 'emol')
+
def test_add_col(self):
impl = self._simple_fixture()
col = Column('g', Integer)
self.metadata.drop_all(self.conn)
self.conn.close()
- def _assert_data(self, data):
+ def _assert_data(self, data, tablename='foo'):
eq_(
- [dict(row) for row in self.conn.execute("select * from foo")],
+ [dict(row) for row
+ in self.conn.execute("select * from %s" % tablename)],
data
)
{"id": 5, "data": "d5", "y": 9}
])
+ def test_rename_column_boolean(self):
+ bar = Table(
+ 'bar', self.metadata,
+ Column('id', Integer, primary_key=True),
+ Column('flag', Boolean()),
+ mysql_engine='InnoDB'
+ )
+ bar.create(self.conn)
+ self.conn.execute(bar.insert(), {'id': 1, 'flag': True})
+ self.conn.execute(bar.insert(), {'id': 2, 'flag': False})
+
+ with self.op.batch_alter_table(
+ "bar"
+ ) as batch_op:
+ batch_op.alter_column(
+ 'flag', new_column_name='bflag', existing_type=Boolean)
+
+ self._assert_data([
+ {"id": 1, 'bflag': True},
+ {"id": 2, 'bflag': False},
+ ], 'bar')
+
+ @config.requirements.non_native_boolean
+ def test_rename_column_non_native_boolean_no_ck(self):
+ bar = Table(
+ 'bar', self.metadata,
+ Column('id', Integer, primary_key=True),
+ Column('flag', Boolean(create_constraint=False)),
+ mysql_engine='InnoDB'
+ )
+ bar.create(self.conn)
+ self.conn.execute(bar.insert(), {'id': 1, 'flag': True})
+ self.conn.execute(bar.insert(), {'id': 2, 'flag': False})
+ self.conn.execute(bar.insert(), {'id': 3, 'flag': 5})
+
+ with self.op.batch_alter_table(
+ "bar",
+ reflect_args=[Column('flag', Boolean(create_constraint=False))]
+ ) as batch_op:
+ batch_op.alter_column(
+ 'flag', new_column_name='bflag', existing_type=Boolean)
+
+ self._assert_data([
+ {"id": 1, 'bflag': True},
+ {"id": 2, 'bflag': False},
+ {'id': 3, 'bflag': 5}
+ ], 'bar')
+
def test_drop_column_pk(self):
with self.op.batch_alter_table("foo") as batch_op:
batch_op.drop_column('id')
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
