if params['referred_schema']:
tname = "%s.%s" % (params['referred_schema'], tname)
+ options = params.get('options', {})
+
const = sa_schema.ForeignKeyConstraint(
[conn_table.c[cname] for cname in params['constrained_columns']],
["%s.%s" % (tname, n) for n in params['referred_columns']],
- onupdate=params.get('onupdate'),
- ondelete=params.get('ondelete'),
- deferrable=params.get('deferrable'),
- initially=params.get('initially'),
+ onupdate=options.get('onupdate'),
+ ondelete=options.get('ondelete'),
+ deferrable=options.get('deferrable'),
+ initially=options.get('initially'),
name=params['name']
)
# needed by 0.7
class _fk_constraint_sig(_constraint_sig):
- def __init__(self, const):
+ def __init__(self, const, include_options=False):
self.const = const
self.name = const.name
- self.source_schema, self.source_table, \
- self.source_columns, self.target_schema, self.target_table, \
- self.target_columns = _fk_spec(const)
+
+ (
+ self.source_schema, self.source_table,
+ self.source_columns, self.target_schema, self.target_table,
+ self.target_columns,
+ onupdate, ondelete,
+ deferrable, initially) = _fk_spec(const)
self.sig = (
self.source_schema, self.source_table, tuple(self.source_columns),
self.target_schema, self.target_table, tuple(self.target_columns)
)
+ if include_options:
+ self.sig += (
+ onupdate.lower() if onupdate else None,
+ ondelete.lower() if ondelete else None,
+ # convert initially + deferrable into one three-state value
+ "initially_deferrable"
+ if initially and initially.lower() == "deferred"
+ else "deferrable" if deferrable
+ else "not deferrable"
+ )
@comparators.dispatch_for("table")
fk for fk in metadata_table.constraints
if isinstance(fk, sa_schema.ForeignKeyConstraint)
)
- metadata_fks = set(_fk_constraint_sig(fk) for fk in metadata_fks)
conn_fks = inspector.get_foreign_keys(tname, schema=schema)
- conn_fks = set(_fk_constraint_sig(_make_foreign_key(const, conn_table))
- for const in conn_fks)
+
+ backend_reflects_fk_options = conn_fks and 'options' in conn_fks[0]
+
+ 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
+ )
conn_fks_by_sig = dict(
(c.sig, c) for c in conn_fks
source_schema, source_table, \
source_columns, target_schema, \
- target_table, target_columns = sqla_compat._fk_spec(constraint)
+ target_table, target_columns,\
+ onupdate, ondelete, deferrable, initially \
+ = sqla_compat._fk_spec(constraint)
kw['source_schema'] = source_schema
kw['referent_schema'] = target_schema
def reflects_pk_names(self):
return exclusions.closed()
+ @property
+ def reflects_fk_options(self):
+ return exclusions.closed()
+
@property
def fail_before_sqla_079(self):
return exclusions.fails_if(
target_schema = constraint.elements[0].column.table.schema
target_table = constraint.elements[0].column.table.name
target_columns = [element.column.name for element in constraint.elements]
-
+ ondelete = constraint.ondelete
+ onupdate = constraint.onupdate
+ deferrable = constraint.deferrable
+ initially = constraint.initially
return (
source_schema, source_table,
- source_columns, target_schema, target_table, target_columns)
+ source_columns, target_schema, target_table, target_columns,
+ onupdate, ondelete, deferrable, initially)
def _is_type_bound(constraint):
Changelog
==========
+.. changelog::
+ :version: 0.8.1
+
+ .. change::
+ :tags: feature, autogenerate
+ :tickets: 317
+
+ Implemented support for autogenerate detection of changes in the
+ ``ondelete``, ``onupdate``, ``initially`` and ``deferrable``
+ attributes of :class:`.ForeignKeyConstraint` objects on
+ SQLAlchemy backends that support these on reflection
+ (as of SQLAlchemy 1.0.8 currently Postgresql for all four,
+ MySQL for ``ondelete`` and ``onupdate`` only). A constraint object
+ that modifies these values will be reported as a "diff" and come out
+ as a drop/create of the constraint with the modified values.
+ The fields are ignored for backends which don't reflect these
+ attributes (as of SQLA 1.0.8 this includes SQLite, Oracle, SQL Server,
+ others).
+
.. changelog::
:version: 0.8.0
:released: August 12, 2015
def _assert_fk_diff(
self, diff, type_, source_table, source_columns,
target_table, target_columns, name=None, conditional_name=None,
- source_schema=None):
+ source_schema=None, onupdate=None, ondelete=None,
+ initially=None, deferrable=None):
# the public API for ForeignKeyConstraint was not very rich
# in 0.7, 0.8, so here we use the well-known but slightly
# private API to get at its elements
(fk_source_schema, fk_source_table,
fk_source_columns, fk_target_schema, fk_target_table,
- fk_target_columns) = _fk_spec(diff[1])
+ fk_target_columns,
+ fk_onupdate, fk_ondelete, fk_deferrable, fk_initially
+ ) = _fk_spec(diff[1])
eq_(diff[0], type_)
eq_(fk_source_table, source_table)
eq_(fk_source_columns, source_columns)
eq_(fk_target_table, target_table)
eq_(fk_source_schema, source_schema)
+ eq_(fk_onupdate, onupdate)
+ eq_(fk_ondelete, ondelete)
+ eq_(fk_initially, initially)
+ eq_(fk_deferrable, deferrable)
eq_([elem.column.name for elem in diff[1].elements],
target_columns)
"""foreign key constraints always have names in the DB"""
return exclusions.fails_on('sqlite')
+ @property
+ def reflects_fk_options(self):
+ return exclusions.only_on(['postgresql', 'mysql'])
+
+ @property
+ def fk_initially(self):
+ """backend supports INITIALLY option in foreign keys"""
+ return exclusions.only_on(['postgresql'])
+
+ @property
+ def fk_deferrable(self):
+ """backend supports DEFERRABLE option in foreign keys"""
+ return exclusions.only_on(['postgresql'])
+
@property
def reflects_unique_constraints_unambiguously(self):
return exclusions.fails_on("mysql")
import sys
-from alembic.testing import TestBase
+from alembic.testing import TestBase, config
from sqlalchemy import MetaData, Column, Table, Integer, String, \
ForeignKeyConstraint
name='fk2'
)
eq_(len(diffs), 2)
+
+
+class AutogenerateFKOptionsTest(AutogenFixtureTest, TestBase):
+ __backend__ = True
+
+ def _fk_opts_fixture(self, old_opts, new_opts):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table('table', m1,
+ Column('id', Integer, primary_key=True),
+ Column('test', String(10)),
+ mysql_engine='InnoDB')
+
+ Table('user', m1,
+ Column('id', Integer, primary_key=True),
+ Column('name', String(50), nullable=False),
+ Column('tid', Integer),
+ ForeignKeyConstraint(['tid'], ['table.id'], **old_opts),
+ mysql_engine='InnoDB')
+
+ Table('table', m2,
+ Column('id', Integer, primary_key=True),
+ Column('test', String(10)),
+ mysql_engine='InnoDB')
+
+ Table('user', m2,
+ Column('id', Integer, primary_key=True),
+ Column('name', String(50), nullable=False),
+ Column('tid', Integer),
+ ForeignKeyConstraint(['tid'], ['table.id'], **new_opts),
+ mysql_engine='InnoDB')
+
+ return self._fixture(m1, m2)
+
+ def _expect_opts_supported(self, deferrable=False, initially=False):
+ if not config.requirements.reflects_fk_options.enabled:
+ return False
+
+ if deferrable and not config.requirements.fk_deferrable.enabled:
+ return False
+
+ if initially and not config.requirements.fk_initially.enabled:
+ return False
+
+ return True
+
+ def test_add_ondelete(self):
+ diffs = self._fk_opts_fixture(
+ {}, {"ondelete": "cascade"}
+ )
+
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ ondelete=None,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ ondelete="cascade"
+ )
+ else:
+ eq_(diffs, [])
+
+ def test_remove_ondelete(self):
+ diffs = self._fk_opts_fixture(
+ {"ondelete": "cascade"}, {}
+ )
+
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ ondelete="CASCADE",
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ ondelete=None
+ )
+ else:
+ eq_(diffs, [])
+
+ def test_nochange_ondelete(self):
+ """test case sensitivity"""
+ diffs = self._fk_opts_fixture(
+ {"ondelete": "caSCAde"}, {"ondelete": "CasCade"}
+ )
+ eq_(diffs, [])
+
+ def test_add_onupdate(self):
+ diffs = self._fk_opts_fixture(
+ {}, {"onupdate": "cascade"}
+ )
+
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate=None,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate="cascade"
+ )
+ else:
+ eq_(diffs, [])
+
+ def test_remove_onupdate(self):
+ diffs = self._fk_opts_fixture(
+ {"onupdate": "cascade"}, {}
+ )
+
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate="CASCADE",
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate=None
+ )
+ else:
+ eq_(diffs, [])
+
+ def test_nochange_onupdate(self):
+ """test case sensitivity"""
+ diffs = self._fk_opts_fixture(
+ {"onupdate": "caSCAde"}, {"onupdate": "CasCade"}
+ )
+ eq_(diffs, [])
+
+ def test_ondelete_onupdate_combo(self):
+ diffs = self._fk_opts_fixture(
+ {"onupdate": "cascade", "ondelete": "set null"},
+ {"onupdate": "restrict", "ondelete": "restrict"}
+ )
+
+ if self._expect_opts_supported():
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate="CASCADE",
+ ondelete="SET NULL",
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ onupdate="restrict",
+ ondelete="restrict"
+ )
+ else:
+ eq_(diffs, [])
+
+ @config.requirements.fk_initially
+ def test_add_initially_deferred(self):
+ diffs = self._fk_opts_fixture(
+ {}, {"initially": "deferred"}
+ )
+
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially=None,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially="deferred"
+ )
+
+ @config.requirements.fk_initially
+ def test_remove_initially_deferred(self):
+ diffs = self._fk_opts_fixture(
+ {"initially": "deferred"}, {}
+ )
+
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially="DEFERRED",
+ deferrable=True,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially=None
+ )
+
+ @config.requirements.fk_deferrable
+ @config.requirements.fk_initially
+ def test_add_initially_immediate_plus_deferrable(self):
+ diffs = self._fk_opts_fixture(
+ {}, {"initially": "immediate", "deferrable": True}
+ )
+
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially=None,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially="immediate",
+ deferrable=True
+ )
+
+ @config.requirements.fk_deferrable
+ @config.requirements.fk_initially
+ def test_remove_initially_immediate_plus_deferrable(self):
+ diffs = self._fk_opts_fixture(
+ {"initially": "immediate", "deferrable": True}, {}
+ )
+
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially=None, # immediate is the default
+ deferrable=True,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ initially=None,
+ deferrable=None
+ )
+
+ @config.requirements.fk_initially
+ @config.requirements.fk_deferrable
+ def test_add_initially_deferrable_nochange_one(self):
+ diffs = self._fk_opts_fixture(
+ {"deferrable": True, "initially": "immediate"},
+ {"deferrable": True, "initially": "immediate"}
+ )
+
+ eq_(diffs, [])
+
+ @config.requirements.fk_initially
+ @config.requirements.fk_deferrable
+ def test_add_initially_deferrable_nochange_two(self):
+ diffs = self._fk_opts_fixture(
+ {"deferrable": True, "initially": "deferred"},
+ {"deferrable": True, "initially": "deferred"}
+ )
+
+ eq_(diffs, [])
+
+ @config.requirements.fk_initially
+ @config.requirements.fk_deferrable
+ def test_add_initially_deferrable_nochange_three(self):
+ diffs = self._fk_opts_fixture(
+ {"deferrable": None, "initially": "deferred"},
+ {"deferrable": None, "initially": "deferred"}
+ )
+
+ eq_(diffs, [])
+
+ @config.requirements.fk_deferrable
+ def test_add_deferrable(self):
+ diffs = self._fk_opts_fixture(
+ {}, {"deferrable": True}
+ )
+
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ deferrable=None,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ deferrable=True
+ )
+
+ @config.requirements.fk_deferrable
+ def test_remove_deferrable(self):
+ diffs = self._fk_opts_fixture(
+ {"deferrable": True}, {}
+ )
+
+ self._assert_fk_diff(
+ diffs[0], "remove_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ deferrable=True,
+ conditional_name="servergenerated"
+ )
+
+ self._assert_fk_diff(
+ diffs[1], "add_fk",
+ "user", ["tid"],
+ "table", ["id"],
+ deferrable=None
+ )
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
