Fix unique together validator doesn't respect condition's fields

rest_framework/serializers.py

@@ -1430,15 +1430,17 @@ def get_unique_together_constraints(self, model):
  """
  for parent_class in [model] + list(model._meta.parents):
  for unique_together in parent_class._meta.unique_together:
- yield unique_together, model._default_manager
+ yield unique_together, model._default_manager, [], None
  for constraint in parent_class._meta.constraints:
  if isinstance(constraint, models.UniqueConstraint) and len(constraint.fields) > 1:
- yield (
- constraint.fields,
- model._default_manager
- if constraint.condition is None
- else model._default_manager.filter(constraint.condition)
- )
+ queryset = model._default_manager
+ if constraint.condition is None:
+ condition_fields = []
+ else:
+ condition_fields = [
+ f[0].split("__")[0] for f in constraint.condition.children
+ ]
+ yield (constraint.fields, queryset, condition_fields, constraint.condition)
 
  def get_uniqueness_extra_kwargs(self, field_names, declared_fields, extra_kwargs):
  """
@@ -1470,9 +1472,9 @@ def get_uniqueness_extra_kwargs(self, field_names, declared_fields, extra_kwargs
 
  # Include each of the `unique_together` and `UniqueConstraint` field names,
  # so long as all the field names are included on the serializer.
- for unique_together_list, queryset in self.get_unique_together_constraints(model):
- if set(field_names).issuperset(unique_together_list):
- unique_constraint_names |= set(unique_together_list)
+ for unique_together_list, queryset, condition_fields, condition in self.get_unique_together_constraints(model):
+ if set(field_names).issuperset((*unique_together_list, *condition_fields)):
+ unique_constraint_names |= set((*unique_together_list, *condition_fields))
 
  # Now we have all the field names that have uniqueness constraints
  # applied, we can add the extra 'required=...' or 'default=...'
@@ -1592,12 +1594,12 @@ def get_unique_together_validators(self):
  # Note that we make sure to check `unique_together` both on the
  # base model class, but also on any parent classes.
  validators = []
- for unique_together, queryset in self.get_unique_together_constraints(self.Meta.model):
+ for unique_together, queryset, condition_fields, condition in self.get_unique_together_constraints(self.Meta.model):
  # Skip if serializer does not map to all unique together sources
- if not set(source_map).issuperset(unique_together):
+ if not set(source_map).issuperset((*unique_together, *condition_fields)):
  continue
 
- for source in unique_together:
+ for source in (*unique_together, *condition_fields):
  assert len(source_map[source]) == 1, (
  "Unable to create `UniqueTogetherValidator` for "
  "`{model}.{field}` as `{serializer}` has multiple "
@@ -1614,9 +1616,9 @@ def get_unique_together_validators(self):
  )
 
  field_names = tuple(source_map[f][0] for f in unique_together)
+ condition_fields = tuple(source_map[f][0] for f in condition_fields)
  validator = UniqueTogetherValidator(
- queryset=queryset,
- fields=field_names
+ queryset=queryset, fields=field_names, condition_fields=condition_fields, condition=condition
  )
  validators.append(validator)
  return validators

rest_framework/validators.py

@@ -7,6 +7,7 @@
 `ModelSerializer` class and an equivalent explicit `Serializer` class.
 """
 from django.db import DataError
+from django.db.models import Exists
 from django.utils.translation import gettext_lazy as _
 
 from rest_framework.exceptions import ValidationError
@@ -22,6 +23,15 @@ def qs_exists(queryset):
  except (TypeError, ValueError, DataError):
  return False
 
+def qs_exists_with_condition(queryset, condition, against):
+ if condition is None:
+ return qs_exists(queryset)
+ try:
+ # use the same query as UniqueConstraint.validate https://github.com/django/django/blob/7ba2a0db20c37a5b1500434ca4ed48022311c171/django/db/models/constraints.py#L672
+ return (condition & Exists(queryset.filter(condition))).check(against)
+ except (TypeError, ValueError, DataError):
+ return False
+
 
 def qs_filter(queryset, **kwargs):
  try:
@@ -99,10 +109,12 @@ class UniqueTogetherValidator:
  missing_message = _('This field is required.')
  requires_context = True
 
- def __init__(self, queryset, fields, message=None):
+ def __init__(self, queryset, fields, message=None, condition_fields=None, condition=None):
  self.queryset = queryset
  self.fields = fields
  self.message = message or self.message
+ self.condition_fields = [] if condition_fields is None else condition_fields
+ self.condition = condition
 
  def enforce_required_fields(self, attrs, serializer):
  """
@@ -114,7 +126,7 @@ def enforce_required_fields(self, attrs, serializer):
 
  missing_items = {
  field_name: self.missing_message
- for field_name in self.fields
+ for field_name in (*self.fields, *self.condition_fields)
  if serializer.fields[field_name].source not in attrs
  }
  if missing_items:
@@ -172,16 +184,22 @@ def __call__(self, attrs, serializer):
  if field in self.fields and value != getattr(serializer.instance, field)
  ]
 
- if checked_values and None not in checked_values and qs_exists(queryset):
+ condition_kwargs = {
+ source: attrs[source]
+ for source in self.condition_fields
+ }
+ if checked_values and None not in checked_values and qs_exists_with_condition(queryset, self.condition, condition_kwargs):
  field_names = ', '.join(self.fields)
  message = self.message.format(field_names=field_names)
  raise ValidationError(message, code='unique')
 
  def __repr__(self):
- return '<%s(queryset=%s, fields=%s)>' % (
+ return '<%s(%s)>' % (
  self.__class__.__name__,
- smart_repr(self.queryset),
- smart_repr(self.fields)
+ ', '.join(
+ f'{attr}={smart_repr(getattr(self, attr))}' 
+ for attr in ('queryset', 'fields', 'condition') 
+ if getattr(self, attr) is not None)
  )
 
  def __eq__(self, other):

tests/test_validators.py

@@ -513,6 +513,11 @@ class Meta:
  name="unique_constraint_model_together_uniq",
  fields=('race_name', 'position'),
  condition=models.Q(race_name='example'),
+ ),
+ models.UniqueConstraint(
+ name="unique_constraint_model_together_uniq2",
+ fields=('race_name', 'position'),
+ condition=models.Q(fancy_conditions__gte=10),
  )
  ]
 
@@ -553,22 +558,56 @@ def test_repr(self):
  position = IntegerField\(.*required=True\)
  global_id = IntegerField\(.*validators=\[<UniqueValidator\(queryset=UniqueConstraintModel.objects.all\(\)\)>\]\)
  class Meta:
- validators = \[<UniqueTogetherValidator\(queryset=<QuerySet \[<UniqueConstraintModel: UniqueConstraintModel object \(1\)>, <UniqueConstraintModel: UniqueConstraintModel object \(2\)>\]>, fields=\('race_name', 'position'\)\)>\]
+ validators = \[<UniqueTogetherValidator\(queryset=UniqueConstraintModel.objects.all\(\), fields=\('race_name', 'position'\), condition=<Q: \(AND: \('race_name', 'example'\)\)>\)>\]
  """)
+ print(repr(serializer))
  assert re.search(expected, repr(serializer)) is not None
 
- def test_unique_together_field(self):
+ def test_unique_together_condition(self):
  """
- UniqueConstraint fields and condition attributes must be passed
- to UniqueTogetherValidator as fields and queryset
+ Fields used in UniqueConstraint's condition must be included
+ into queryset existence check
  """
- serializer = UniqueConstraintSerializer()
- assert len(serializer.validators) == 1
- validator = serializer.validators[0]
- assert validator.fields == ('race_name', 'position')
- assert set(validator.queryset.values_list(flat=True)) == set(
- UniqueConstraintModel.objects.filter(race_name='example').values_list(flat=True)
+ UniqueConstraintModel.objects.create(
+ race_name='condition',
+ position=1,
+ global_id=10,
+ fancy_conditions=10
  )
+ serializer = UniqueConstraintSerializer(data={
+ 'race_name': 'condition',
+ 'position': 1,
+ 'global_id': 11,
+ 'fancy_conditions': 9,
+ })
+ assert serializer.is_valid()
+ serializer = UniqueConstraintSerializer(data={
+ 'race_name': 'condition',
+ 'position': 1,
+ 'global_id': 11,
+ 'fancy_conditions': 11,
+ })
+ assert not serializer.is_valid()
+
+ def test_unique_together_condition_fields_required(self):
+ """
+ Fields used in UniqueConstraint's condition must be present in serializer
+ """
+ serializer = UniqueConstraintSerializer(data={
+ 'race_name': 'condition',
+ 'position': 1,
+ 'global_id': 11,
+ })
+ assert not serializer.is_valid()
+ assert serializer.errors == {'fancy_conditions': ['This field is required.']}
+
+ class NoFieldsSerializer(serializers.ModelSerializer):
+ class Meta:
+ model = UniqueConstraintModel
+ fields = ('race_name', 'position', 'global_id')
+
+ serializer = NoFieldsSerializer()
+ assert len(serializer.validators) == 1
 
  def test_single_field_uniq_validators(self):
  """
@@ -579,7 +618,7 @@ def test_single_field_uniq_validators(self):
  extra_validators_qty = 2 if django_version[0] >= 5 else 0
  #
  serializer = UniqueConstraintSerializer()
- assert len(serializer.validators) == 1
+ assert len(serializer.validators) == 2
  validators = serializer.fields['global_id'].validators
  assert len(validators) == 1 + extra_validators_qty
  assert validators[0].queryset == UniqueConstraintModel.objects