from sqlalchemy import schema as sql_schema
from sqlalchemy.util import OrderedDict
from .. import util
+if util.sqla_08:
+ from sqlalchemy.events import SchemaEventTarget
from ..util.sqla_compat import _columns_for_constraint, \
_is_type_bound, _fk_is_self_referential
for c in self.table.c:
c_copy = c.copy(schema=schema)
c_copy.unique = c_copy.index = False
+ # ensure that the type object was copied,
+ # as we may need to modify it in-place
+ if isinstance(c.type, SchemaEventTarget):
+ assert c_copy.type is not c.type
self.columns[c.name] = c_copy
self.named_constraints = {}
self.unnamed_constraints = []
for const in self.table.constraints:
if _is_type_bound(const):
continue
- if const.name:
+ elif const.name:
self.named_constraints[const.name] = const
else:
self.unnamed_constraints.append(const)
if type_ is not None:
type_ = sqltypes.to_instance(type_)
+ # old type is being discarded so turn off eventing
+ # rules. Alternatively we can
+ # erase the events set up by this type, but this is simpler.
+ # we also ignore the drop_constraint that will come here from
+ # Operations.implementation_for(alter_column)
+ if isinstance(existing.type, SchemaEventTarget):
+ existing.type._create_events = \
+ existing.type.create_constraint = False
+
existing.type = type_
+
+ # we *dont* however set events for the new type, because
+ # alter_column is invoked from
+ # Operations.implementation_for(alter_column) which already
+ # will emit an add_constraint()
+
existing_transfer["expr"] = cast(existing_transfer["expr"], type_)
if nullable is not None:
existing.nullable = nullable
try:
del self.named_constraints[const.name]
except KeyError:
+ if const._type_bound:
+ # type-bound constraints are only included in the new
+ # table via their type object in any case, so ignore the
+ # drop_constraint() that comes here via the
+ # Operations.implementation_for(alter_column)
+ return
raise ValueError("No such constraint: '%s'" % const.name)
def create_index(self, idx):
):
batch_op.add_column(Column('foo', Integer))
+Changing the Type of Boolean, Enum and other implicit CHECK datatypes
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The SQLAlchemy types :class:`~sqlalchemy.types.Boolean` and
+:class:`~sqlalchemy.types.Enum` are part of a category of types known as
+"schema" types; this style of type creates other structures along with the
+type itself, most commonly (but not always) a CHECK constraint.
+
+Alembic handles dropping and creating the CHECK constraints here automatically,
+including in the case of batch mode. When changing the type of an existing
+column, what's necessary is that the existing type be specified fully::
+
+ with self.op.batch_alter_table("some_table"):
+ batch_op.alter_column(
+ 'q', type_=Integer,
+ existing_type=Boolean(create_constraint=True, constraint_name="ck1"))
+
Including CHECK constraints
^^^^^^^^^^^^^^^^^^^^^^^^^^^
batch_op.add_column(Column('foo', Integer))
batch_op.drop_column('bar')
+Note this only includes CHECK constraints that are explicitly stated
+as part of the table definition, not the CHECK constraints that are generated
+by datatypes such as :class:`~sqlalchemy.types.Boolean` or
+:class:`~sqlalchemy.types.Enum`.
Dealing with Referencing Foreign Keys
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
if isinstance(const, CheckConstraint)]),
1)
+ def test_change_type_schematype_to_non(self):
+ impl = self._boolean_fixture()
+ impl.alter_column('tname', 'flag', type_=Integer)
+ new_table = self._assert_impl(
+ impl, colnames=['id', 'flag'],
+ ddl_not_contains="CHECK")
+ assert new_table.c.flag.type._type_affinity is Integer
+
+ # NOTE: we can't do test_change_type_non_to_schematype
+ # at this level because the "add_constraint" part of this
+ # comes from toimpl.py, which we aren't testing here
+
def test_rename_col_boolean_no_ck(self):
impl = self._boolean_no_ck_fixture()
impl.alter_column('tname', 'flag', name='bflag')
'ALTER TABLE _alembic_batch_temp RENAME TO foo'
)
+ def test_change_type_from_schematype(self):
+ context = self._fixture()
+ self.table.append_column(
+ Column('y', Boolean(
+ create_constraint=True, name="ck1")))
+
+ with self.op.batch_alter_table(
+ "foo", copy_from=self.table) as batch_op:
+ batch_op.alter_column(
+ 'y', type_=Integer,
+ existing_type=Boolean(
+ create_constraint=True, name="ck1"))
+ context.assert_(
+ 'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+ 'data VARCHAR(50), x INTEGER, y INTEGER, PRIMARY KEY (id))',
+ 'INSERT INTO _alembic_batch_temp (id, data, x, y) SELECT foo.id, '
+ 'foo.data, foo.x, CAST(foo.y AS INTEGER) AS anon_1 FROM foo',
+ 'DROP TABLE foo',
+ 'ALTER TABLE _alembic_batch_temp RENAME TO foo'
+ )
+
+ def test_change_type_to_schematype(self):
+ context = self._fixture()
+ self.table.append_column(
+ Column('y', Integer))
+
+ with self.op.batch_alter_table(
+ "foo", copy_from=self.table) as batch_op:
+ batch_op.alter_column(
+ 'y', existing_type=Integer,
+ type_=Boolean(
+ create_constraint=True, name="ck1"))
+ context.assert_(
+ 'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
+ 'data VARCHAR(50), x INTEGER, y BOOLEAN, PRIMARY KEY (id), '
+ 'CONSTRAINT ck1 CHECK (y IN (0, 1)))',
+ 'INSERT INTO _alembic_batch_temp (id, data, x, y) SELECT foo.id, '
+ 'foo.data, foo.x, CAST(foo.y AS BOOLEAN) AS anon_1 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(
t.create(self.conn)
return t
+ def _boolean_fixture(self):
+ t = Table(
+ 'hasbool', self.metadata,
+ Column('x', Boolean(create_constraint=True, name='ck1')),
+ Column('y', Integer)
+ )
+ t.create(self.conn)
+
+ def _int_to_boolean_fixture(self):
+ t = Table(
+ 'hasbool', self.metadata,
+ Column('x', Integer)
+ )
+ t.create(self.conn)
+
+ def test_change_type_boolean_to_int(self):
+ self._boolean_fixture()
+ with self.op.batch_alter_table(
+ "hasbool"
+ ) as batch_op:
+ batch_op.alter_column(
+ 'x', type_=Integer, existing_type=Boolean(
+ create_constraint=True, name='ck1'))
+ insp = Inspector.from_engine(config.db)
+
+ eq_(
+ [c['type']._type_affinity for c in insp.get_columns('hasbool')
+ if c['name'] == 'x'],
+ [Integer]
+ )
+
+ def test_drop_col_schematype(self):
+ self._boolean_fixture()
+ with self.op.batch_alter_table(
+ "hasbool"
+ ) as batch_op:
+ batch_op.drop_column('x')
+ insp = Inspector.from_engine(config.db)
+
+ assert 'x' not in (c['name'] for c in insp.get_columns('hasbool'))
+
+ def test_change_type_int_to_boolean(self):
+ self._int_to_boolean_fixture()
+ with self.op.batch_alter_table(
+ "hasbool"
+ ) as batch_op:
+ batch_op.alter_column(
+ 'x', type_=Boolean(create_constraint=True, name='ck1'))
+ insp = Inspector.from_engine(config.db)
+
+ if exclusions.against(config, "sqlite"):
+ eq_(
+ [c['type']._type_affinity for
+ c in insp.get_columns('hasbool') if c['name'] == 'x'],
+ [Boolean]
+ )
+ elif exclusions.against(config, "mysql"):
+ eq_(
+ [c['type']._type_affinity for
+ c in insp.get_columns('hasbool') if c['name'] == 'x'],
+ [Integer]
+ )
+
def tearDown(self):
self.metadata.drop_all(self.conn)
self.conn.close()
def test_create_drop_index(self):
super(BatchRoundTripPostgresqlTest, self).test_create_drop_index()
+
+ @exclusions.fails()
+ def test_change_type_int_to_boolean(self):
+ super(BatchRoundTripPostgresqlTest, self).\
+ test_change_type_int_to_boolean()
+
+ @exclusions.fails()
+ def test_change_type_boolean_to_int(self):
+ super(BatchRoundTripPostgresqlTest, self).\
+ test_change_type_boolean_to_int()
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
