from os import path
-__version__ = '0.8.1'
+__version__ = '0.8.2'
package_dir = path.abspath(path.dirname(__file__))
)
if include_options:
self.sig += (
- onupdate.lower() if onupdate else None,
- ondelete.lower() if ondelete else None,
+ (None if onupdate.lower() == 'no action'
+ else onupdate.lower())
+ if onupdate else None,
+ (None if ondelete.lower() == 'no action'
+ else ondelete.lower())
+ if ondelete else None,
# convert initially + deferrable into one three-state value
"initially_deferrable"
if initially and initially.lower() == "deferred"
backend_reflects_fk_options = conn_fks and 'options' in conn_fks[0]
+ conn_fks = set(_make_foreign_key(const, conn_table) for const in conn_fks)
+
+ # give the dialect a chance to correct the FKs to match more
+ # closely
+ autogen_context.migration_context.impl.\
+ correct_for_autogen_foreignkeys(
+ conn_fks, metadata_fks,
+ )
+
metadata_fks = set(
_fk_constraint_sig(fk, include_options=backend_reflects_fk_options)
for fk in metadata_fks
)
conn_fks = set(
- _fk_constraint_sig(
- _make_foreign_key(const, conn_table),
- include_options=backend_reflects_fk_options
- )
- for const in conn_fks
+ _fk_constraint_sig(fk, include_options=backend_reflects_fk_options)
+ for fk in conn_fks
)
conn_fks_by_sig = dict(
inspector, table, column_info)
return adapt
+ def correct_for_autogen_foreignkeys(self, conn_fks, metadata_fks):
+ pass
+
def autogen_column_reflect(self, inspector, table, column_info):
"""A hook that is attached to the 'column_reflect' event for when
a Table is reflected from the database during the autogenerate
elif overlap not in metadata_ix_names:
conn_indexes.discard(conn_ix_names[overlap])
+ def correct_for_autogen_foreignkeys(self, conn_fks, metadata_fks):
+ conn_fk_by_sig = dict(
+ (compare._fk_constraint_sig(fk).sig, fk) for fk in conn_fks
+ )
+ metadata_fk_by_sig = dict(
+ (compare._fk_constraint_sig(fk).sig, fk) for fk in metadata_fks
+ )
+
+ for sig in set(conn_fk_by_sig).intersection(metadata_fk_by_sig):
+ mdfk = metadata_fk_by_sig[sig]
+ cnfk = conn_fk_by_sig[sig]
+ # MySQL considers RESTRICT to be the default and doesn't
+ # report on it. if the model has explicit RESTRICT and
+ # the conn FK has None, set it to RESTRICT
+ if mdfk.ondelete is not None and \
+ mdfk.ondelete.lower() == 'restrict' and \
+ cnfk.ondelete is None:
+ cnfk.ondelete = 'RESTRICT'
+ if mdfk.onupdate is not None and \
+ mdfk.onupdate.lower() == 'restrict' and \
+ cnfk.onupdate is None:
+ cnfk.onupdate = 'RESTRICT'
+
class MySQLAlterDefault(AlterColumn):
Changelog
==========
+.. changelog::
+ :version: 0.8.2
+
+ .. change::
+ :tags: bug, autogenerate
+ :tickets: 321
+
+ Added workaround in new foreign key option detection feature for
+ MySQL's consideration of the "RESTRICT" option being the default,
+ for which no value is reported from the database; the MySQL impl now
+ corrects for when the model reports RESTRICT but the database reports
+ nothing. A similar rule is in the default FK comparison to accommodate
+ for the default "NO ACTION" setting being present in the model but not
+ necessarily reported by the database, or vice versa.
+
.. changelog::
:version: 0.8.1
:released: August 22, 2015
import sys
-from alembic.testing import TestBase, config
+from alembic.testing import TestBase, config, mock
from sqlalchemy import MetaData, Column, Table, Integer, String, \
ForeignKeyConstraint
)
eq_(diffs, [])
+ def test_nochange_ondelete_restrict(self):
+ """test the RESTRICT option which MySQL doesn't report on"""
+
+ diffs = self._fk_opts_fixture(
+ {"ondelete": "restrict"}, {"ondelete": "restrict"}
+ )
+ eq_(diffs, [])
+
+ def test_nochange_onupdate_restrict(self):
+ """test the RESTRICT option which MySQL doesn't report on"""
+
+ diffs = self._fk_opts_fixture(
+ {"onupdate": "restrict"}, {"onupdate": "restrict"}
+ )
+ eq_(diffs, [])
+
+ def test_nochange_ondelete_noaction(self):
+ """test the NO ACTION option which generally comes back as None"""
+
+ diffs = self._fk_opts_fixture(
+ {"ondelete": "no action"}, {"ondelete": "no action"}
+ )
+ eq_(diffs, [])
+
+ def test_nochange_onupdate_noaction(self):
+ """test the NO ACTION option which generally comes back as None"""
+
+ diffs = self._fk_opts_fixture(
+ {"onupdate": "no action"}, {"onupdate": "no action"}
+ )
+ eq_(diffs, [])
+
+ def test_change_ondelete_from_restrict(self):
+ """test the RESTRICT option which MySQL doesn't report on"""
+
+ # note that this is impossible to detect if we change
+ # from RESTRICT to NO ACTION on MySQL.
+ diffs = self._fk_opts_fixture(
+ {"ondelete": "restrict"}, {"ondelete": "cascade"}
+ )
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate=None,
+ ondelete=mock.ANY, # MySQL reports None, PG reports RESTRICT
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate=None,
+ ondelete="cascade"
+ )
+ else:
+ eq_(diffs, [])
+
+ def test_change_onupdate_from_restrict(self):
+ """test the RESTRICT option which MySQL doesn't report on"""
+
+ # note that this is impossible to detect if we change
+ # from RESTRICT to NO ACTION on MySQL.
+ diffs = self._fk_opts_fixture(
+ {"onupdate": "restrict"}, {"onupdate": "cascade"}
+ )
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate=mock.ANY, # MySQL reports None, PG reports RESTRICT
+ ondelete=None,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate="cascade",
+ ondelete=None
+ )
+ else:
+ eq_(diffs, [])
+
def test_ondelete_onupdate_combo(self):
diffs = self._fk_opts_fixture(
{"onupdate": "cascade", "ondelete": "set null"},
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
