Fixed #20581 -- Added support for deferrable unique constraints.

This commit is contained in:
Ian Foote 2018-08-27 03:25:06 +01:00 committed by Mariusz Felisiak
parent 555e3a848e
commit c226c6cb32
14 changed files with 457 additions and 16 deletions

View File

@ -20,6 +20,8 @@ class BaseDatabaseFeatures:
# Does the backend allow inserting duplicate rows when a unique_together # Does the backend allow inserting duplicate rows when a unique_together
# constraint exists and some fields are nullable but not all of them? # constraint exists and some fields are nullable but not all of them?
supports_partially_nullable_unique_constraints = True supports_partially_nullable_unique_constraints = True
# Does the backend support initially deferrable unique constraints?
supports_deferrable_unique_constraints = False
can_use_chunked_reads = True can_use_chunked_reads = True
can_return_columns_from_insert = False can_return_columns_from_insert = False

View File

@ -5,7 +5,7 @@ from django.db.backends.ddl_references import (
Columns, ForeignKeyName, IndexName, Statement, Table, Columns, ForeignKeyName, IndexName, Statement, Table,
) )
from django.db.backends.utils import names_digest, split_identifier from django.db.backends.utils import names_digest, split_identifier
from django.db.models import Index from django.db.models import Deferrable, Index
from django.db.transaction import TransactionManagementError, atomic from django.db.transaction import TransactionManagementError, atomic
from django.utils import timezone from django.utils import timezone
@ -65,7 +65,7 @@ class BaseDatabaseSchemaEditor:
sql_rename_column = "ALTER TABLE %(table)s RENAME COLUMN %(old_column)s TO %(new_column)s" sql_rename_column = "ALTER TABLE %(table)s RENAME COLUMN %(old_column)s TO %(new_column)s"
sql_update_with_default = "UPDATE %(table)s SET %(column)s = %(default)s WHERE %(column)s IS NULL" sql_update_with_default = "UPDATE %(table)s SET %(column)s = %(default)s WHERE %(column)s IS NULL"
sql_unique_constraint = "UNIQUE (%(columns)s)" sql_unique_constraint = "UNIQUE (%(columns)s)%(deferrable)s"
sql_check_constraint = "CHECK (%(check)s)" sql_check_constraint = "CHECK (%(check)s)"
sql_delete_constraint = "ALTER TABLE %(table)s DROP CONSTRAINT %(name)s" sql_delete_constraint = "ALTER TABLE %(table)s DROP CONSTRAINT %(name)s"
sql_constraint = "CONSTRAINT %(name)s %(constraint)s" sql_constraint = "CONSTRAINT %(name)s %(constraint)s"
@ -73,7 +73,7 @@ class BaseDatabaseSchemaEditor:
sql_create_check = "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s CHECK (%(check)s)" sql_create_check = "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s CHECK (%(check)s)"
sql_delete_check = sql_delete_constraint sql_delete_check = sql_delete_constraint
sql_create_unique = "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s UNIQUE (%(columns)s)" sql_create_unique = "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s UNIQUE (%(columns)s)%(deferrable)s"
sql_delete_unique = sql_delete_constraint sql_delete_unique = sql_delete_constraint
sql_create_fk = ( sql_create_fk = (
@ -1075,7 +1075,20 @@ class BaseDatabaseSchemaEditor:
def _delete_fk_sql(self, model, name): def _delete_fk_sql(self, model, name):
return self._delete_constraint_sql(self.sql_delete_fk, model, name) return self._delete_constraint_sql(self.sql_delete_fk, model, name)
def _unique_sql(self, model, fields, name, condition=None): def _deferrable_constraint_sql(self, deferrable):
if deferrable is None:
return ''
if deferrable == Deferrable.DEFERRED:
return ' DEFERRABLE INITIALLY DEFERRED'
if deferrable == Deferrable.IMMEDIATE:
return ' DEFERRABLE INITIALLY IMMEDIATE'
def _unique_sql(self, model, fields, name, condition=None, deferrable=None):
if (
deferrable and
not self.connection.features.supports_deferrable_unique_constraints
):
return None
if condition: if condition:
# Databases support conditional unique constraints via a unique # Databases support conditional unique constraints via a unique
# index. # index.
@ -1085,13 +1098,20 @@ class BaseDatabaseSchemaEditor:
return None return None
constraint = self.sql_unique_constraint % { constraint = self.sql_unique_constraint % {
'columns': ', '.join(map(self.quote_name, fields)), 'columns': ', '.join(map(self.quote_name, fields)),
'deferrable': self._deferrable_constraint_sql(deferrable),
} }
return self.sql_constraint % { return self.sql_constraint % {
'name': self.quote_name(name), 'name': self.quote_name(name),
'constraint': constraint, 'constraint': constraint,
} }
def _create_unique_sql(self, model, columns, name=None, condition=None): def _create_unique_sql(self, model, columns, name=None, condition=None, deferrable=None):
if (
deferrable and
not self.connection.features.supports_deferrable_unique_constraints
):
return None
def create_unique_name(*args, **kwargs): def create_unique_name(*args, **kwargs):
return self.quote_name(self._create_index_name(*args, **kwargs)) return self.quote_name(self._create_index_name(*args, **kwargs))
@ -1113,9 +1133,15 @@ class BaseDatabaseSchemaEditor:
name=name, name=name,
columns=columns, columns=columns,
condition=self._index_condition_sql(condition), condition=self._index_condition_sql(condition),
deferrable=self._deferrable_constraint_sql(deferrable),
) )
def _delete_unique_sql(self, model, name, condition=None): def _delete_unique_sql(self, model, name, condition=None, deferrable=None):
if (
deferrable and
not self.connection.features.supports_deferrable_unique_constraints
):
return None
if condition: if condition:
return ( return (
self._delete_constraint_sql(self.sql_delete_index, model, name) self._delete_constraint_sql(self.sql_delete_index, model, name)

View File

@ -71,9 +71,17 @@ def wrap_oracle_errors():
# message = 'ORA-02091: transaction rolled back # message = 'ORA-02091: transaction rolled back
# 'ORA-02291: integrity constraint (TEST_DJANGOTEST.SYS # 'ORA-02291: integrity constraint (TEST_DJANGOTEST.SYS
# _C00102056) violated - parent key not found' # _C00102056) violated - parent key not found'
# or:
# 'ORA-00001: unique constraint (DJANGOTEST.DEFERRABLE_
# PINK_CONSTRAINT) violated
# Convert that case to Django's IntegrityError exception. # Convert that case to Django's IntegrityError exception.
x = e.args[0] x = e.args[0]
if hasattr(x, 'code') and hasattr(x, 'message') and x.code == 2091 and 'ORA-02291' in x.message: if (
hasattr(x, 'code') and
hasattr(x, 'message') and
x.code == 2091 and
('ORA-02291' in x.message or 'ORA-00001' in x.message)
):
raise IntegrityError(*tuple(e.args)) raise IntegrityError(*tuple(e.args))
raise raise

View File

@ -17,6 +17,7 @@ class DatabaseFeatures(BaseDatabaseFeatures):
has_native_duration_field = True has_native_duration_field = True
can_defer_constraint_checks = True can_defer_constraint_checks = True
supports_partially_nullable_unique_constraints = False supports_partially_nullable_unique_constraints = False
supports_deferrable_unique_constraints = True
truncates_names = True truncates_names = True
supports_tablespaces = True supports_tablespaces = True
supports_sequence_reset = False supports_sequence_reset = False

View File

@ -56,6 +56,7 @@ class DatabaseFeatures(BaseDatabaseFeatures):
supports_aggregate_filter_clause = True supports_aggregate_filter_clause = True
supported_explain_formats = {'JSON', 'TEXT', 'XML', 'YAML'} supported_explain_formats = {'JSON', 'TEXT', 'XML', 'YAML'}
validates_explain_options = False # A query will error on invalid options. validates_explain_options = False # A query will error on invalid options.
supports_deferrable_unique_constraints = True
@cached_property @cached_property
def is_postgresql_9_6(self): def is_postgresql_9_6(self):

View File

@ -1904,6 +1904,25 @@ class Model(metaclass=ModelBase):
id='models.W036', id='models.W036',
) )
) )
if not (
connection.features.supports_deferrable_unique_constraints or
'supports_deferrable_unique_constraints' in cls._meta.required_db_features
) and any(
isinstance(constraint, UniqueConstraint) and constraint.deferrable is not None
for constraint in cls._meta.constraints
):
errors.append(
checks.Warning(
'%s does not support deferrable unique constraints.'
% connection.display_name,
hint=(
"A constraint won't be created. Silence this "
"warning if you don't care about it."
),
obj=cls,
id='models.W038',
)
)
return errors return errors

View File

@ -1,7 +1,9 @@
from enum import Enum
from django.db.models.query_utils import Q from django.db.models.query_utils import Q
from django.db.models.sql.query import Query from django.db.models.sql.query import Query
__all__ = ['CheckConstraint', 'UniqueConstraint'] __all__ = ['CheckConstraint', 'Deferrable', 'UniqueConstraint']
class BaseConstraint: class BaseConstraint:
@ -69,14 +71,28 @@ class CheckConstraint(BaseConstraint):
return path, args, kwargs return path, args, kwargs
class Deferrable(Enum):
DEFERRED = 'deferred'
IMMEDIATE = 'immediate'
class UniqueConstraint(BaseConstraint): class UniqueConstraint(BaseConstraint):
def __init__(self, *, fields, name, condition=None): def __init__(self, *, fields, name, condition=None, deferrable=None):
if not fields: if not fields:
raise ValueError('At least one field is required to define a unique constraint.') raise ValueError('At least one field is required to define a unique constraint.')
if not isinstance(condition, (type(None), Q)): if not isinstance(condition, (type(None), Q)):
raise ValueError('UniqueConstraint.condition must be a Q instance.') raise ValueError('UniqueConstraint.condition must be a Q instance.')
if condition and deferrable:
raise ValueError(
'UniqueConstraint with conditions cannot be deferred.'
)
if not isinstance(deferrable, (type(None), Deferrable)):
raise ValueError(
'UniqueConstraint.deferrable must be a Deferrable instance.'
)
self.fields = tuple(fields) self.fields = tuple(fields)
self.condition = condition self.condition = condition
self.deferrable = deferrable
super().__init__(name) super().__init__(name)
def _get_condition_sql(self, model, schema_editor): def _get_condition_sql(self, model, schema_editor):
@ -91,21 +107,30 @@ class UniqueConstraint(BaseConstraint):
def constraint_sql(self, model, schema_editor): def constraint_sql(self, model, schema_editor):
fields = [model._meta.get_field(field_name).column for field_name in self.fields] fields = [model._meta.get_field(field_name).column for field_name in self.fields]
condition = self._get_condition_sql(model, schema_editor) condition = self._get_condition_sql(model, schema_editor)
return schema_editor._unique_sql(model, fields, self.name, condition=condition) return schema_editor._unique_sql(
model, fields, self.name, condition=condition,
deferrable=self.deferrable,
)
def create_sql(self, model, schema_editor): def create_sql(self, model, schema_editor):
fields = [model._meta.get_field(field_name).column for field_name in self.fields] fields = [model._meta.get_field(field_name).column for field_name in self.fields]
condition = self._get_condition_sql(model, schema_editor) condition = self._get_condition_sql(model, schema_editor)
return schema_editor._create_unique_sql(model, fields, self.name, condition=condition) return schema_editor._create_unique_sql(
model, fields, self.name, condition=condition,
deferrable=self.deferrable,
)
def remove_sql(self, model, schema_editor): def remove_sql(self, model, schema_editor):
condition = self._get_condition_sql(model, schema_editor) condition = self._get_condition_sql(model, schema_editor)
return schema_editor._delete_unique_sql(model, self.name, condition=condition) return schema_editor._delete_unique_sql(
model, self.name, condition=condition, deferrable=self.deferrable,
)
def __repr__(self): def __repr__(self):
return '<%s: fields=%r name=%r%s>' % ( return '<%s: fields=%r name=%r%s%s>' % (
self.__class__.__name__, self.fields, self.name, self.__class__.__name__, self.fields, self.name,
'' if self.condition is None else ' condition=%s' % self.condition, '' if self.condition is None else ' condition=%s' % self.condition,
'' if self.deferrable is None else ' deferrable=%s' % self.deferrable,
) )
def __eq__(self, other): def __eq__(self, other):
@ -113,7 +138,8 @@ class UniqueConstraint(BaseConstraint):
return ( return (
self.name == other.name and self.name == other.name and
self.fields == other.fields and self.fields == other.fields and
self.condition == other.condition self.condition == other.condition and
self.deferrable == other.deferrable
) )
return super().__eq__(other) return super().__eq__(other)
@ -122,4 +148,6 @@ class UniqueConstraint(BaseConstraint):
kwargs['fields'] = self.fields kwargs['fields'] = self.fields
if self.condition: if self.condition:
kwargs['condition'] = self.condition kwargs['condition'] = self.condition
if self.deferrable:
kwargs['deferrable'] = self.deferrable
return path, args, kwargs return path, args, kwargs

View File

@ -354,6 +354,8 @@ Models
* **models.W036**: ``<database>`` does not support unique constraints with * **models.W036**: ``<database>`` does not support unique constraints with
conditions. conditions.
* **models.W037**: ``<database>`` does not support indexes with conditions. * **models.W037**: ``<database>`` does not support indexes with conditions.
* **models.W038**: ``<database>`` does not support deferrable unique
constraints.
Security Security
-------- --------

View File

@ -76,7 +76,7 @@ The name of the constraint.
``UniqueConstraint`` ``UniqueConstraint``
==================== ====================
.. class:: UniqueConstraint(*, fields, name, condition=None) .. class:: UniqueConstraint(*, fields, name, condition=None, deferrable=None)
Creates a unique constraint in the database. Creates a unique constraint in the database.
@ -119,3 +119,35 @@ ensures that each user only has one draft.
These conditions have the same database restrictions as These conditions have the same database restrictions as
:attr:`Index.condition`. :attr:`Index.condition`.
``deferrable``
--------------
.. attribute:: UniqueConstraint.deferrable
.. versionadded:: 3.1
Set this parameter to create a deferrable unique constraint. Accepted values
are ``Deferrable.DEFERRED`` or ``Deferrable.IMMEDIATE``. For example::
from django.db.models import Deferrable, UniqueConstraint
UniqueConstraint(
name='unique_order',
fields=['order'],
deferrable=Deferrable.DEFERRED,
)
By default constraints are not deferred. A deferred constraint will not be
enforced until the end of the transaction. An immediate constraint will be
enforced immediately after every command.
.. admonition:: MySQL, MariaDB, and SQLite.
Deferrable unique constraints are ignored on MySQL, MariaDB, and SQLite as
neither supports them.
.. warning::
Deferred unique constraints may lead to a `performance penalty
<https://www.postgresql.org/docs/current/sql-createtable.html#id-1.9.3.85.9.4>`_.

View File

@ -381,6 +381,9 @@ Models
<sqlite3.Connection.create_function>` on Python 3.8+. This allows using them <sqlite3.Connection.create_function>` on Python 3.8+. This allows using them
in check constraints and partial indexes. in check constraints and partial indexes.
* The new :attr:`.UniqueConstraint.deferrable` attribute allows creating
deferrable unique constraints.
Pagination Pagination
~~~~~~~~~~ ~~~~~~~~~~

View File

@ -59,6 +59,28 @@ class UniqueConstraintConditionProduct(models.Model):
] ]
class UniqueConstraintDeferrable(models.Model):
name = models.CharField(max_length=255)
shelf = models.CharField(max_length=31)
class Meta:
required_db_features = {
'supports_deferrable_unique_constraints',
}
constraints = [
models.UniqueConstraint(
fields=['name'],
name='name_init_deferred_uniq',
deferrable=models.Deferrable.DEFERRED,
),
models.UniqueConstraint(
fields=['shelf'],
name='sheld_init_immediate_uniq',
deferrable=models.Deferrable.IMMEDIATE,
),
]
class AbstractModel(models.Model): class AbstractModel(models.Model):
age = models.IntegerField() age = models.IntegerField()

View File

@ -3,11 +3,12 @@ from unittest import mock
from django.core.exceptions import ValidationError from django.core.exceptions import ValidationError
from django.db import IntegrityError, connection, models from django.db import IntegrityError, connection, models
from django.db.models.constraints import BaseConstraint from django.db.models.constraints import BaseConstraint
from django.db.transaction import atomic
from django.test import SimpleTestCase, TestCase, skipUnlessDBFeature from django.test import SimpleTestCase, TestCase, skipUnlessDBFeature
from .models import ( from .models import (
ChildModel, Product, UniqueConstraintConditionProduct, ChildModel, Product, UniqueConstraintConditionProduct,
UniqueConstraintProduct, UniqueConstraintDeferrable, UniqueConstraintProduct,
) )
@ -166,6 +167,20 @@ class UniqueConstraintTests(TestCase):
), ),
) )
def test_eq_with_deferrable(self):
constraint_1 = models.UniqueConstraint(
fields=['foo', 'bar'],
name='unique',
deferrable=models.Deferrable.DEFERRED,
)
constraint_2 = models.UniqueConstraint(
fields=['foo', 'bar'],
name='unique',
deferrable=models.Deferrable.IMMEDIATE,
)
self.assertEqual(constraint_1, constraint_1)
self.assertNotEqual(constraint_1, constraint_2)
def test_repr(self): def test_repr(self):
fields = ['foo', 'bar'] fields = ['foo', 'bar']
name = 'unique_fields' name = 'unique_fields'
@ -187,6 +202,18 @@ class UniqueConstraintTests(TestCase):
"condition=(AND: ('foo', F(bar)))>", "condition=(AND: ('foo', F(bar)))>",
) )
def test_repr_with_deferrable(self):
constraint = models.UniqueConstraint(
fields=['foo', 'bar'],
name='unique_fields',
deferrable=models.Deferrable.IMMEDIATE,
)
self.assertEqual(
repr(constraint),
"<UniqueConstraint: fields=('foo', 'bar') name='unique_fields' "
"deferrable=Deferrable.IMMEDIATE>",
)
def test_deconstruction(self): def test_deconstruction(self):
fields = ['foo', 'bar'] fields = ['foo', 'bar']
name = 'unique_fields' name = 'unique_fields'
@ -206,6 +233,23 @@ class UniqueConstraintTests(TestCase):
self.assertEqual(args, ()) self.assertEqual(args, ())
self.assertEqual(kwargs, {'fields': tuple(fields), 'name': name, 'condition': condition}) self.assertEqual(kwargs, {'fields': tuple(fields), 'name': name, 'condition': condition})
def test_deconstruction_with_deferrable(self):
fields = ['foo']
name = 'unique_fields'
constraint = models.UniqueConstraint(
fields=fields,
name=name,
deferrable=models.Deferrable.DEFERRED,
)
path, args, kwargs = constraint.deconstruct()
self.assertEqual(path, 'django.db.models.UniqueConstraint')
self.assertEqual(args, ())
self.assertEqual(kwargs, {
'fields': tuple(fields),
'name': name,
'deferrable': models.Deferrable.DEFERRED,
})
def test_database_constraint(self): def test_database_constraint(self):
with self.assertRaises(IntegrityError): with self.assertRaises(IntegrityError):
UniqueConstraintProduct.objects.create(name=self.p1.name, color=self.p1.color) UniqueConstraintProduct.objects.create(name=self.p1.name, color=self.p1.color)
@ -238,3 +282,54 @@ class UniqueConstraintTests(TestCase):
def test_condition_must_be_q(self): def test_condition_must_be_q(self):
with self.assertRaisesMessage(ValueError, 'UniqueConstraint.condition must be a Q instance.'): with self.assertRaisesMessage(ValueError, 'UniqueConstraint.condition must be a Q instance.'):
models.UniqueConstraint(name='uniq', fields=['name'], condition='invalid') models.UniqueConstraint(name='uniq', fields=['name'], condition='invalid')
@skipUnlessDBFeature('supports_deferrable_unique_constraints')
def test_initially_deferred_database_constraint(self):
obj_1 = UniqueConstraintDeferrable.objects.create(name='p1', shelf='front')
obj_2 = UniqueConstraintDeferrable.objects.create(name='p2', shelf='back')
def swap():
obj_1.name, obj_2.name = obj_2.name, obj_1.name
obj_1.save()
obj_2.save()
swap()
# Behavior can be changed with SET CONSTRAINTS.
with self.assertRaises(IntegrityError):
with atomic(), connection.cursor() as cursor:
constraint_name = connection.ops.quote_name('name_init_deferred_uniq')
cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % constraint_name)
swap()
@skipUnlessDBFeature('supports_deferrable_unique_constraints')
def test_initially_immediate_database_constraint(self):
obj_1 = UniqueConstraintDeferrable.objects.create(name='p1', shelf='front')
obj_2 = UniqueConstraintDeferrable.objects.create(name='p2', shelf='back')
obj_1.shelf, obj_2.shelf = obj_2.shelf, obj_1.shelf
with self.assertRaises(IntegrityError), atomic():
obj_1.save()
# Behavior can be changed with SET CONSTRAINTS.
with connection.cursor() as cursor:
constraint_name = connection.ops.quote_name('sheld_init_immediate_uniq')
cursor.execute('SET CONSTRAINTS %s DEFERRED' % constraint_name)
obj_1.save()
obj_2.save()
def test_deferrable_with_condition(self):
message = 'UniqueConstraint with conditions cannot be deferred.'
with self.assertRaisesMessage(ValueError, message):
models.UniqueConstraint(
fields=['name'],
name='name_without_color_unique',
condition=models.Q(color__isnull=True),
deferrable=models.Deferrable.DEFERRED,
)
def test_invalid_defer_argument(self):
message = 'UniqueConstraint.deferrable must be a Deferrable instance.'
with self.assertRaisesMessage(ValueError, message):
models.UniqueConstraint(
fields=['name'],
name='name_invalid',
deferrable='invalid',
)

View File

@ -1414,3 +1414,47 @@ class ConstraintsTests(TestCase):
] ]
self.assertEqual(Model.check(databases=self.databases), []) self.assertEqual(Model.check(databases=self.databases), [])
def test_deferrable_unique_constraint(self):
class Model(models.Model):
age = models.IntegerField()
class Meta:
constraints = [
models.UniqueConstraint(
fields=['age'],
name='unique_age_deferrable',
deferrable=models.Deferrable.DEFERRED,
),
]
errors = Model.check(databases=self.databases)
expected = [] if connection.features.supports_deferrable_unique_constraints else [
Warning(
'%s does not support deferrable unique constraints.'
% connection.display_name,
hint=(
"A constraint won't be created. Silence this warning if "
"you don't care about it."
),
obj=Model,
id='models.W038',
),
]
self.assertEqual(errors, expected)
def test_deferrable_unique_constraint_required_db_features(self):
class Model(models.Model):
age = models.IntegerField()
class Meta:
required_db_features = {'supports_deferrable_unique_constraints'}
constraints = [
models.UniqueConstraint(
fields=['age'],
name='unique_age_deferrable',
deferrable=models.Deferrable.IMMEDIATE,
),
]
self.assertEqual(Model.check(databases=self.databases), [])

View File

@ -393,6 +393,60 @@ class OperationTests(OperationTestBase):
self.assertEqual(definition[1], []) self.assertEqual(definition[1], [])
self.assertEqual(definition[2]['options']['constraints'], [partial_unique_constraint]) self.assertEqual(definition[2]['options']['constraints'], [partial_unique_constraint])
def test_create_model_with_deferred_unique_constraint(self):
deferred_unique_constraint = models.UniqueConstraint(
fields=['pink'],
name='deferrable_pink_constraint',
deferrable=models.Deferrable.DEFERRED,
)
operation = migrations.CreateModel(
'Pony',
[
('id', models.AutoField(primary_key=True)),
('pink', models.IntegerField(default=3)),
],
options={'constraints': [deferred_unique_constraint]},
)
project_state = ProjectState()
new_state = project_state.clone()
operation.state_forwards('test_crmo', new_state)
self.assertEqual(len(new_state.models['test_crmo', 'pony'].options['constraints']), 1)
self.assertTableNotExists('test_crmo_pony')
# Create table.
with connection.schema_editor() as editor:
operation.database_forwards('test_crmo', editor, project_state, new_state)
self.assertTableExists('test_crmo_pony')
Pony = new_state.apps.get_model('test_crmo', 'Pony')
Pony.objects.create(pink=1)
if connection.features.supports_deferrable_unique_constraints:
# Unique constraint is deferred.
with transaction.atomic():
obj = Pony.objects.create(pink=1)
obj.pink = 2
obj.save()
# Constraint behavior can be changed with SET CONSTRAINTS.
with self.assertRaises(IntegrityError):
with transaction.atomic(), connection.cursor() as cursor:
quoted_name = connection.ops.quote_name(deferred_unique_constraint.name)
cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % quoted_name)
obj = Pony.objects.create(pink=1)
obj.pink = 3
obj.save()
else:
Pony.objects.create(pink=1)
# Reversal.
with connection.schema_editor() as editor:
operation.database_backwards('test_crmo', editor, new_state, project_state)
self.assertTableNotExists('test_crmo_pony')
# Deconstruction.
definition = operation.deconstruct()
self.assertEqual(definition[0], 'CreateModel')
self.assertEqual(definition[1], [])
self.assertEqual(
definition[2]['options']['constraints'],
[deferred_unique_constraint],
)
def test_create_model_managers(self): def test_create_model_managers(self):
""" """
The managers on a model are set. The managers on a model are set.
@ -2046,6 +2100,110 @@ class OperationTests(OperationTestBase):
'name': 'test_constraint_pony_pink_for_weight_gt_5_uniq', 'name': 'test_constraint_pony_pink_for_weight_gt_5_uniq',
}) })
def test_add_deferred_unique_constraint(self):
app_label = 'test_adddeferred_uc'
project_state = self.set_up_test_model(app_label)
deferred_unique_constraint = models.UniqueConstraint(
fields=['pink'],
name='deferred_pink_constraint_add',
deferrable=models.Deferrable.DEFERRED,
)
operation = migrations.AddConstraint('Pony', deferred_unique_constraint)
self.assertEqual(
operation.describe(),
'Create constraint deferred_pink_constraint_add on model Pony',
)
# Add constraint.
new_state = project_state.clone()
operation.state_forwards(app_label, new_state)
self.assertEqual(len(new_state.models[app_label, 'pony'].options['constraints']), 1)
Pony = new_state.apps.get_model(app_label, 'Pony')
self.assertEqual(len(Pony._meta.constraints), 1)
with connection.schema_editor() as editor:
operation.database_forwards(app_label, editor, project_state, new_state)
Pony.objects.create(pink=1, weight=4.0)
if connection.features.supports_deferrable_unique_constraints:
# Unique constraint is deferred.
with transaction.atomic():
obj = Pony.objects.create(pink=1, weight=4.0)
obj.pink = 2
obj.save()
# Constraint behavior can be changed with SET CONSTRAINTS.
with self.assertRaises(IntegrityError):
with transaction.atomic(), connection.cursor() as cursor:
quoted_name = connection.ops.quote_name(deferred_unique_constraint.name)
cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % quoted_name)
obj = Pony.objects.create(pink=1, weight=4.0)
obj.pink = 3
obj.save()
else:
Pony.objects.create(pink=1, weight=4.0)
# Reversal.
with connection.schema_editor() as editor:
operation.database_backwards(app_label, editor, new_state, project_state)
# Constraint doesn't work.
Pony.objects.create(pink=1, weight=4.0)
# Deconstruction.
definition = operation.deconstruct()
self.assertEqual(definition[0], 'AddConstraint')
self.assertEqual(definition[1], [])
self.assertEqual(
definition[2],
{'model_name': 'Pony', 'constraint': deferred_unique_constraint},
)
def test_remove_deferred_unique_constraint(self):
app_label = 'test_removedeferred_uc'
deferred_unique_constraint = models.UniqueConstraint(
fields=['pink'],
name='deferred_pink_constraint_rm',
deferrable=models.Deferrable.DEFERRED,
)
project_state = self.set_up_test_model(app_label, constraints=[deferred_unique_constraint])
operation = migrations.RemoveConstraint('Pony', deferred_unique_constraint.name)
self.assertEqual(
operation.describe(),
'Remove constraint deferred_pink_constraint_rm from model Pony',
)
# Remove constraint.
new_state = project_state.clone()
operation.state_forwards(app_label, new_state)
self.assertEqual(len(new_state.models[app_label, 'pony'].options['constraints']), 0)
Pony = new_state.apps.get_model(app_label, 'Pony')
self.assertEqual(len(Pony._meta.constraints), 0)
with connection.schema_editor() as editor:
operation.database_forwards(app_label, editor, project_state, new_state)
# Constraint doesn't work.
Pony.objects.create(pink=1, weight=4.0)
Pony.objects.create(pink=1, weight=4.0).delete()
# Reversal.
with connection.schema_editor() as editor:
operation.database_backwards(app_label, editor, new_state, project_state)
if connection.features.supports_deferrable_unique_constraints:
# Unique constraint is deferred.
with transaction.atomic():
obj = Pony.objects.create(pink=1, weight=4.0)
obj.pink = 2
obj.save()
# Constraint behavior can be changed with SET CONSTRAINTS.
with self.assertRaises(IntegrityError):
with transaction.atomic(), connection.cursor() as cursor:
quoted_name = connection.ops.quote_name(deferred_unique_constraint.name)
cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % quoted_name)
obj = Pony.objects.create(pink=1, weight=4.0)
obj.pink = 3
obj.save()
else:
Pony.objects.create(pink=1, weight=4.0)
# Deconstruction.
definition = operation.deconstruct()
self.assertEqual(definition[0], 'RemoveConstraint')
self.assertEqual(definition[1], [])
self.assertEqual(definition[2], {
'model_name': 'Pony',
'name': 'deferred_pink_constraint_rm',
})
def test_alter_model_options(self): def test_alter_model_options(self):
""" """
Tests the AlterModelOptions operation. Tests the AlterModelOptions operation.