using the ultimate table name, rather than ``_alembic_batch_temp``.
When the table is renamed from ``_alembic_batch_temp`` back to the
original name, the FK now points to the right name. This
will **not** work if referential integrity is being enforced (eg. SQLite
"PRAGMA FOREIGN_KEYS=ON") since the original table is dropped and
the new table then renamed to that name, however this is now consistent
with how foreign key constraints on **other** tables already operate
with batch mode; these don't support batch mode if referential integrity
is enabled in any case.
fixes #345
from sqlalchemy import schema as sql_schema
from sqlalchemy.util import OrderedDict
from .. import util
-from ..util.sqla_compat import _columns_for_constraint, _is_type_bound
+from ..util.sqla_compat import _columns_for_constraint, \
+ _is_type_bound, _fk_is_self_referential
class BatchOperationsImpl(object):
if not const_columns.issubset(self.column_transfers):
continue
- const_copy = const.copy(schema=schema, target_table=new_table)
+
+ if isinstance(const, ForeignKeyConstraint):
+ if _fk_is_self_referential(const):
+ # for self-referential constraint, refer to the
+ # *original* table name, and not _alembic_batch_temp.
+ # This is consistent with how we're handling
+ # FK constraints from other tables; we assume SQLite
+ # no foreign keys just keeps the names unchanged, so
+ # when we rename back, they match again.
+ const_copy = const.copy(
+ schema=schema, target_table=self.table)
+ else:
+ # "target_table" for ForeignKeyConstraint.copy() is
+ # only used if the FK is detected as being
+ # self-referential, which we are handling above.
+ const_copy = const.copy(schema=schema)
+ else:
+ const_copy = const.copy(schema=schema, target_table=new_table)
if isinstance(const, ForeignKeyConstraint):
self._setup_referent(m, const)
new_table.append_constraint(const_copy)
referent_schema = parts[0]
else:
referent_schema = None
+
if tname != '_alembic_batch_temp':
key = sql_schema._get_table_key(tname, referent_schema)
if key in metadata.tables:
return skip_if(BooleanPredicate(True, "marked as skip"))
-def fails():
- return fails_if(BooleanPredicate(True, "expected to fail"))
+def fails(msg=None):
+ return fails_if(BooleanPredicate(True, msg or "expected to fail"))
@decorator
onupdate, ondelete, deferrable, initially)
+def _fk_is_self_referential(constraint):
+ spec = constraint.elements[0]._get_colspec()
+ tokens = spec.split(".")
+ tokens.pop(-1) # colname
+ tablekey = ".".join(tokens)
+ return tablekey == constraint.parent.key
+
+
def _is_type_bound(constraint):
# this deals with SQLAlchemy #3260, don't copy CHECK constraints
# that will be generated by the type.
Dealing with Referencing Foreign Keys
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-If the SQLite database is enforcing referential integrity with
-``PRAGMA FOREIGN KEYS``, this pragma may need to be disabled when the workflow
-mode proceeds, else remote constraints which refer to this table may prevent
-it from being dropped; additionally, for referential integrity to be
-re-enabled, it may be necessary to recreate the
-foreign keys on those remote tables to refer again to the new table (this
-is definitely the case on other databases, at least). SQLite is normally used
-without referential integrity enabled so this won't be a problem for most
-users.
+It is important to note that batch table operations **do not work** with
+foreign keys that enforce referential integrity. This because the
+target table is dropped; if foreign keys refer to it, this will raise
+an error. On SQLite, whether or not foreign keys actually enforce is
+controlled by the ``PRAGMA FOREIGN KEYS`` pragma; this pragma, if in use,
+must be disabled when the workflow mode proceeds. When the operation is
+complete, the batch-migrated table will have the same name
+that it started with, so those referring foreign keys will again
+refer to this table.
+
+A special case is dealing with self-referring foreign keys. Here,
+Alembic takes a special step of recreating the self-referring foreign key
+as referring to the original table name, rather than at the "temp" table,
+so that like in the case of other foreign key constraints, when the table
+is renamed to its original name, the foreign key
+again references the correct table. This operation only works when
+referential integrity is disabled, consistent with the same requirement
+for referring foreign keys from other tables.
+
+.. versionchanged:: 0.8.4 Self-referring foreign keys are created with the
+ target table name in batch mode, even though this table will temporarily
+ not exist when dropped. This requires that the target database is not
+ enforcing referential integrity.
+
+When SQLite's ``PRAGMA FOREIGN KEYS`` mode is turned on, it does provide
+the service that foreign key constraints, including self-referential, will
+automatically be modified to point to their table across table renames,
+however this mode prevents the target table from being dropped as is required
+by a batch migration. Therefore it may be necessary to manipulate the
+``PRAGMA FOREIGN KEYS`` setting if a migration seeks to rename a table vs.
+batch migrate it.
.. _batch_offline_mode:
.. changelog::
:version: 0.8.4
+ .. change::
+ :tags: bug, batch
+ :tickets: 345
+
+ Batch mode generates a FOREIGN KEY constraint that is self-referential
+ using the ultimate table name, rather than ``_alembic_batch_temp``.
+ When the table is renamed from ``_alembic_batch_temp`` back to the
+ original name, the FK now points to the right name. This
+ will **not** work if referential integrity is being enforced (eg. SQLite
+ "PRAGMA FOREIGN_KEYS=ON") since the original table is dropped and
+ the new table then renamed to that name, however this is now consistent
+ with how foreign key constraints on **other** tables already operate
+ with batch mode; these don't support batch mode if referential integrity
+ is enabled in any case.
+
.. change::
:tags: bug, autogenerate
:tickets: 341
context = MigrationContext.configure(self.conn)
self.op = Operations(context)
+ @contextmanager
+ def _sqlite_referential_integrity(self):
+ self.conn.execute("PRAGMA foreign_keys=ON")
+ try:
+ yield
+ finally:
+ self.conn.execute("PRAGMA foreign_keys=OFF")
+
def _no_pk_fixture(self):
nopk = Table(
'nopk', self.metadata,
def test_fk_points_to_me_recreate(self):
self._test_fk_points_to_me("always")
+ @exclusions.only_on("sqlite")
+ @exclusions.fails(
+ "intentionally asserting that this "
+ "doesn't work w/ pragma foreign keys")
+ def test_fk_points_to_me_sqlite_refinteg(self):
+ with self._sqlite_referential_integrity():
+ self._test_fk_points_to_me("auto")
+
def _test_fk_points_to_me(self, recreate):
bar = Table(
'bar', self.metadata,
batch_op.alter_column(
'data', new_column_name='newdata', existing_type=String(50))
+ insp = Inspector.from_engine(self.conn)
+ eq_(
+ [(key['referred_table'],
+ key['referred_columns'], key['constrained_columns'])
+ for key in insp.get_foreign_keys('bar')],
+ [('foo', ['id'], ['foo_id'])]
+ )
+
+ def test_selfref_fk_auto(self):
+ self._test_selfref_fk("auto")
+
+ @config.requirements.no_referential_integrity
+ def test_selfref_fk_recreate(self):
+ self._test_selfref_fk("always")
+
+ @exclusions.only_on("sqlite")
+ @exclusions.fails(
+ "intentionally asserting that this "
+ "doesn't work w/ pragma foreign keys")
+ def test_selfref_fk_sqlite_refinteg(self):
+ with self._sqlite_referential_integrity():
+ self._test_selfref_fk("auto")
+
+ def _test_selfref_fk(self, recreate):
+ bar = Table(
+ 'bar', self.metadata,
+ Column('id', Integer, primary_key=True),
+ Column('bar_id', Integer, ForeignKey('bar.id')),
+ Column('data', String(50)),
+ mysql_engine='InnoDB'
+ )
+ bar.create(self.conn)
+ self.conn.execute(bar.insert(), {'id': 1, 'data': 'x', 'bar_id': None})
+ self.conn.execute(bar.insert(), {'id': 2, 'data': 'y', 'bar_id': 1})
+
+ with self.op.batch_alter_table("bar", recreate=recreate) as batch_op:
+ batch_op.alter_column(
+ 'data', new_column_name='newdata', existing_type=String(50))
+
+ insp = Inspector.from_engine(self.conn)
+
+ insp = Inspector.from_engine(self.conn)
+ eq_(
+ [(key['referred_table'],
+ key['referred_columns'], key['constrained_columns'])
+ for key in insp.get_foreign_keys('bar')],
+ [('bar', ['id'], ['bar_id'])]
+ )
+
def test_change_type(self):
with self.op.batch_alter_table("foo") as batch_op:
batch_op.alter_column('data', type_=Integer)
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
