--- /dev/null
+.. change::
+ :tags: usecase, postgresql
+ :tickets: 11595
+
+ Added support for specifying a list of columns for ``SET NULL`` and ``SET
+ DEFAULT`` actions of ``ON DELETE`` clause of foreign key definition on
+ PostgreSQL. Pull request courtesy Denis Laxalde.
+
+ .. seealso::
+
+ :ref:`postgresql_constraint_options`
),
)
-Note that these clauses require ``InnoDB`` tables when used with MySQL.
-They may also not be supported on other databases.
+Note that some backends have special requirements for cascades to function:
+
+* MySQL / MariaDB - the ``InnoDB`` storage engine should be used (this is
+ typically the default in modern databases)
+* SQLite - constraints are not enabled by default.
+ See :ref:`sqlite_foreign_keys`
.. seealso::
:ref:`passive_deletes_many_to_many`
+ :ref:`postgresql_constraint_options` - indicates additional options
+ available for foreign key cascades such as column lists
+
+ :ref:`sqlite_foreign_keys` - background on enabling foreign key support
+ with SQLite
+
.. _schema_unique_constraint:
UNIQUE Constraint
<https://www.postgresql.org/docs/current/static/sql-altertable.html>`_ -
in the PostgreSQL documentation.
+* Column list with foreign key ``ON DELETE SET`` actions: This applies to
+ :class:`.ForeignKey` and :class:`.ForeignKeyConstraint`, the :paramref:`.ForeignKey.ondelete`
+ parameter will accept on the PostgreSQL backend only a string list of column
+ names inside parenthesis, following the ``SET NULL`` or ``SET DEFAULT``
+ phrases, which will limit the set of columns that are subject to the
+ action::
+
+ fktable = Table(
+ "fktable",
+ metadata,
+ Column("tid", Integer),
+ Column("id", Integer),
+ Column("fk_id_del_set_null", Integer),
+ ForeignKeyConstraint(
+ columns=["tid", "fk_id_del_set_null"],
+ refcolumns=[pktable.c.tid, pktable.c.id],
+ ondelete="SET NULL (fk_id_del_set_null)",
+ ),
+ )
+
+ .. versionadded:: 2.0.40
+
+
.. _postgresql_table_valued_overview:
Table values, Table and Column valued functions, Row and Tuple objects
"verbose",
}
+
colspecs = {
sqltypes.ARRAY: _array.ARRAY,
sqltypes.Interval: INTERVAL,
text += self._define_constraint_validity(constraint)
return text
+ @util.memoized_property
+ def _fk_ondelete_pattern(self):
+ return re.compile(
+ r"^(?:RESTRICT|CASCADE|SET (?:NULL|DEFAULT)(?:\s*\(.+\))?"
+ r"|NO ACTION)$",
+ re.I,
+ )
+
+ def define_constraint_ondelete_cascade(self, constraint):
+ return " ON DELETE %s" % self.preparer.validate_sql_phrase(
+ constraint.ondelete, self._fk_ondelete_pattern
+ )
+
def visit_create_enum_type(self, create, **kw):
type_ = create.element
r"[\s]?(ON UPDATE "
r"(CASCADE|RESTRICT|NO ACTION|SET NULL|SET DEFAULT)+)?"
r"[\s]?(ON DELETE "
- r"(CASCADE|RESTRICT|NO ACTION|SET NULL|SET DEFAULT)+)?"
+ r"(CASCADE|RESTRICT|NO ACTION|"
+ r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
r"[\s]?(DEFERRABLE|NOT DEFERRABLE)?"
r"[\s]?(INITIALLY (DEFERRED|IMMEDIATE)+)?"
)
) -> str:
text = ""
if constraint.ondelete is not None:
- text += " ON DELETE %s" % self.preparer.validate_sql_phrase(
- constraint.ondelete, FK_ON_DELETE
- )
+ text += self.define_constraint_ondelete_cascade(constraint)
+
if constraint.onupdate is not None:
- text += " ON UPDATE %s" % self.preparer.validate_sql_phrase(
- constraint.onupdate, FK_ON_UPDATE
- )
+ text += self.define_constraint_onupdate_cascade(constraint)
return text
+ def define_constraint_ondelete_cascade(
+ self, constraint: ForeignKeyConstraint
+ ) -> str:
+ return " ON DELETE %s" % self.preparer.validate_sql_phrase(
+ constraint.ondelete, FK_ON_DELETE
+ )
+
+ def define_constraint_onupdate_cascade(
+ self, constraint: ForeignKeyConstraint
+ ) -> str:
+ return " ON UPDATE %s" % self.preparer.validate_sql_phrase(
+ constraint.onupdate, FK_ON_UPDATE
+ )
+
def define_constraint_deferrability(self, constraint: Constraint) -> str:
text = ""
if constraint.deferrable is not None:
issuing DDL for this constraint. Typical values include CASCADE,
DELETE and RESTRICT.
+ .. seealso::
+
+ :ref:`on_update_on_delete`
+
:param ondelete: Optional string. If set, emit ON DELETE <value> when
issuing DDL for this constraint. Typical values include CASCADE,
- SET NULL and RESTRICT.
+ SET NULL and RESTRICT. Some dialects may allow for additional
+ syntaxes.
+
+ .. seealso::
+
+ :ref:`on_update_on_delete`
:param deferrable: Optional bool. If set, emit DEFERRABLE or NOT
DEFERRABLE when issuing DDL for this constraint.
:param name: Optional, the in-database name of the key.
:param onupdate: Optional string. If set, emit ON UPDATE <value> when
- issuing DDL for this constraint. Typical values include CASCADE,
- DELETE and RESTRICT.
+ issuing DDL for this constraint. Typical values include CASCADE,
+ DELETE and RESTRICT.
+
+ .. seealso::
+
+ :ref:`on_update_on_delete`
:param ondelete: Optional string. If set, emit ON DELETE <value> when
- issuing DDL for this constraint. Typical values include CASCADE,
- SET NULL and RESTRICT.
+ issuing DDL for this constraint. Typical values include CASCADE,
+ SET NULL and RESTRICT. Some dialects may allow for additional
+ syntaxes.
+
+ .. seealso::
+
+ :ref:`on_update_on_delete`
:param deferrable: Optional bool. If set, emit DEFERRABLE or NOT
DEFERRABLE when issuing DDL for this constraint.
")",
)
+ def test_create_foreign_key_constraint_ondelete_column_list(self):
+ m = MetaData()
+ pktable = Table(
+ "pktable",
+ m,
+ Column("tid", Integer, primary_key=True),
+ Column("id", Integer, primary_key=True),
+ )
+ fktable = Table(
+ "fktable",
+ m,
+ Column("tid", Integer),
+ Column("id", Integer),
+ Column("fk_id_del_set_null", Integer),
+ Column("fk_id_del_set_default", Integer, server_default=text("0")),
+ ForeignKeyConstraint(
+ columns=["tid", "fk_id_del_set_null"],
+ refcolumns=[pktable.c.tid, pktable.c.id],
+ ondelete="SET NULL (fk_id_del_set_null)",
+ ),
+ ForeignKeyConstraint(
+ columns=["tid", "fk_id_del_set_default"],
+ refcolumns=[pktable.c.tid, pktable.c.id],
+ ondelete="SET DEFAULT(fk_id_del_set_default)",
+ ),
+ )
+
+ self.assert_compile(
+ schema.CreateTable(fktable),
+ "CREATE TABLE fktable ("
+ "tid INTEGER, id INTEGER, "
+ "fk_id_del_set_null INTEGER, "
+ "fk_id_del_set_default INTEGER DEFAULT 0, "
+ "FOREIGN KEY(tid, fk_id_del_set_null)"
+ " REFERENCES pktable (tid, id)"
+ " ON DELETE SET NULL (fk_id_del_set_null), "
+ "FOREIGN KEY(tid, fk_id_del_set_default)"
+ " REFERENCES pktable (tid, id)"
+ " ON DELETE SET DEFAULT(fk_id_del_set_default)"
+ ")",
+ )
+
def test_exclude_constraint_min(self):
m = MetaData()
tbl = Table("testtbl", m, Column("room", Integer, primary_key=True))
from sqlalchemy import Column
from sqlalchemy import exc
from sqlalchemy import ForeignKey
+from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import inspect
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy import Text
+from sqlalchemy import text
from sqlalchemy import UniqueConstraint
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.dialects.postgresql import base as postgresql
subject = Table("subject", meta2, autoload_with=connection)
eq_(subject.primary_key.columns.keys(), ["p2", "p1"])
+ def test_reflected_foreign_key_ondelete_column_list(
+ self, metadata, connection
+ ):
+ meta1 = metadata
+ pktable = Table(
+ "pktable",
+ meta1,
+ Column("tid", Integer, primary_key=True),
+ Column("id", Integer, primary_key=True),
+ )
+ Table(
+ "fktable",
+ meta1,
+ Column("tid", Integer),
+ Column("id", Integer),
+ Column("fk_id_del_set_null", Integer),
+ Column("fk_id_del_set_default", Integer, server_default=text("0")),
+ ForeignKeyConstraint(
+ name="fktable_tid_fk_id_del_set_null_fkey",
+ columns=["tid", "fk_id_del_set_null"],
+ refcolumns=[pktable.c.tid, pktable.c.id],
+ ondelete="SET NULL (fk_id_del_set_null)",
+ ),
+ ForeignKeyConstraint(
+ name="fktable_tid_fk_id_del_set_default_fkey",
+ columns=["tid", "fk_id_del_set_default"],
+ refcolumns=[pktable.c.tid, pktable.c.id],
+ ondelete="SET DEFAULT(fk_id_del_set_default)",
+ ),
+ )
+
+ meta1.create_all(connection)
+ meta2 = MetaData()
+ fktable = Table("fktable", meta2, autoload_with=connection)
+ fkey_set_null = next(
+ c
+ for c in fktable.foreign_key_constraints
+ if c.name == "fktable_tid_fk_id_del_set_null_fkey"
+ )
+ eq_(fkey_set_null.ondelete, "SET NULL (fk_id_del_set_null)")
+ fkey_set_default = next(
+ c
+ for c in fktable.foreign_key_constraints
+ if c.name == "fktable_tid_fk_id_del_set_default_fkey"
+ )
+ eq_(fkey_set_default.ondelete, "SET DEFAULT (fk_id_del_set_default)")
+
def test_pg_weirdchar_reflection(self, metadata, connection):
meta1 = metadata
subject = Table(
import datetime
import decimal
+import re
from typing import TYPE_CHECKING
from sqlalchemy import alias
"FOO RESTRICT",
"CASCADE WRONG",
"SET NULL",
+ # test that PostgreSQL's syntax added in #11595 is not
+ # accepted by base compiler
+ "SET NULL(postgresql_db.some_column)",
):
const = schema.AddConstraint(
schema.ForeignKeyConstraint(
)
assert_raises_message(
exc.CompileError,
- r"Unexpected SQL phrase: '%s'" % phrase,
+ rf"Unexpected SQL phrase: '{re.escape(phrase)}'",
const.compile,
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
