Django中UniqueConstraint约束下更新收藏字段报错求助

数据老陈

2026-6-23

Django UniqueConstraint 更新触发唯一约束冲突的解决方法

我在使用Django的UniqueConstraint功能时遇到更新问题：业务规则要求每个用户仅能拥有1条收藏查询，因此在QueryModel中配置了条件唯一约束，指定user字段+is_favorite=True的组合唯一，约束名为unique_favorite_per_user。但在QueryView视图集的partial_update方法中尝试将某条查询设为收藏时，触发IntegrityError，提示违反该唯一约束。希望实现设置某条查询为收藏时，自动将当前用户其他查询的is_favorite设为False，无需手动遍历检查。

相关代码

模型代码

class QueryModel(models.Model):

    id = models.AutoField(_("query id"), primary_key=True)
    user = models.ForeignKey(UserModel, on_delete=models.CASCADE)
    name = models.CharField(_("query name"), max_length=150)
    status = models.CharField(
        _("query status"),
        choices=QueryStatus.choices,
        default=QueryStatus.ACTIVE,
    )
    is_favorite = models.BooleanField(_("favorite"), default=False)
    date_created = models.DateTimeField(_("date created"), auto_now_add=True)

    class Meta(object):
        app_label = "app"
        db_table = "query"
        constraints = [
            models.UniqueConstraint(
                fields=("user",),
                condition=Q(is_favorite=True),
                name="unique_favorite_per_user",
            )
        ]

视图集代码

class QueryView(viewsets.ModelViewSet):
    """
    View to list all users in the system.

    * Requires token authentication.
    * Only admin users are able to access this view.
    """

    authentication_classes = [TokenAuthentication]
    permission_classes = [IsAuthenticated]
    serializer_class = QuerySerializer
    aws_client = AWSClient()

    def get_queryset(self):
        """
        This view should return a list of all the purchases
        for the currently authenticated user.
        """
        self._paginator = None
        return QueryModel.objects.filter(user=self.request.user).order_by("id")

    def create(self, request):
        serializer = QuerySerializer(
            data={
                "user": request.user.id,
                "name": request.data["name"],
                "expiration_date": request.data["expiration_date"],
                "json": request.data["json"],
            }
        )

        serializer.is_valid(raise_exception=True)
        serializer.save()

        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def partial_update(self, request, pk=None):
        serializer = QuerySerializer(data=request.data, partial=True)
        serializer.is_valid(raise_exception=True)

        if "is_favorite" in serializer.validated_data.keys():
            self.get_queryset().update(
                is_favorite=ExpressionWrapper(Q(pk=pk), BooleanField())
            )

        return super().partial_update(request, pk)

URL配置代码

router = DefaultRouter(trailing_slash=False)
router.register("auth", AuthView, basename="auth")
router.register("query", QueryView, basename="query")

version_info = json.load(open("./version.json"))


@api_view(["GET"])
def health_check(request):
    """Check if the endpoint is on. Used by container orchestration system."""
    return Response({"status": "pass", **version_info})


urlpatterns = [
    path("", health_check),
    *router.urls,
]

报错信息

Traceback (most recent call last):
   File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 105, in _execute
     return self.cursor.execute(sql, params)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "unique_favorite_per_user"
 DETAIL:  Key (user_workspace_id)=(2) already exists.


 The above exception was the direct cause of the following exception:

 Traceback (most recent call last):
   File "/usr/local/lib/python3.11/site-packages/django/core/handlers/exception.py", line 55, in inner
     response = get_response(request)
                ^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/core/handlers/base.py", line 197, in _get_response
     response = wrapped_callback(request, *callback_args, **callback_kwargs)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/views/decorators/csrf.py", line 65, in _view_wrapper
     return view_func(request, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/rest_framework/viewsets.py", line 124, in view
     return self.dispatch(request, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/rest_framework/views.py", line 509, in dispatch
     response = self.handle_exception(exc)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/rest_framework/views.py", line 469, in handle_exception
     self.raise_uncaught_exception(exc)
   File "/usr/local/lib/python3.11/site-packages/rest_framework/views.py", line 480, in raise_uncaught_exception
     raise exc
   File "/usr/local/lib/python3.11/site-packages/rest_framework/views.py", line 506, in dispatch
     response = handler(request, *args, **kwargs)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/var/lib/trends_backend/trends/views/query.py", line 61, in partial_update
     self.get_queryset().update(
   File "/usr/local/lib/python3.11/site-packages/django/db/models/query.py", line 1253, in update
     rows = query.get_compiler(self.db).execute_sql(CURSOR)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/db/models/sql/compiler.py", line 1990, in execute_sql
     cursor = super().execute_sql(result_type)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/db/models/sql/compiler.py", line 1562, in execute_sql
     cursor.execute(sql, params)
   File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 122, in execute
     return super().execute(sql, params)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 79, in execute
     return self._execute_with_wrappers(
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 92, in _execute_with_wrappers
     return executor(sql, params, many, context)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 100, in _execute
     with self.db.wrap_database_errors:
   File "/usr/local/lib/python3.11/site-packages/django/db/utils.py", line 91, in __exit__
     raise dj_exc_value.with_traceback(traceback) from exc_value
   File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 105, in _execute
     return self.cursor.execute(sql, params)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 django.db.utils.IntegrityError: duplicate key value violates unique constraint "unique_favorite_per_user"
 DETAIL:  Key (user_id)=(2) already exists.

问题原因与解决方案

问题根源

原代码中使用self.get_queryset().update(is_favorite=ExpressionWrapper(Q(pk=pk), BooleanField()))尝试批量更新所有行，但数据库执行该SQL时，会同时修改所有行的is_favorite值，中间会出现「旧收藏项未改为False，新项已改为True」的瞬间状态，直接触发唯一约束校验失败。此外，原代码的序列化器未传入实例，且后续调用super().partial_update会导致重复更新。

解决方案1：修改视图集的partial_update方法

from django.db import transaction

def partial_update(self, request, pk=None):
    instance = self.get_object()
    serializer = QuerySerializer(instance, data=request.data, partial=True)
    serializer.is_valid(raise_exception=True)

    if "is_favorite" in serializer.validated_data and serializer.validated_data["is_favorite"]:
        # 原子事务确保两步更新同时生效，无中间状态
        with transaction.atomic():
            # 先将当前用户其他所有查询的收藏状态设为False
            self.get_queryset().exclude(pk=pk).update(is_favorite=False)
            # 再设置当前查询为收藏
            instance.is_favorite = True
            instance.save()
        return Response(serializer.data, status=status.HTTP_200_OK)
    
    # 非收藏设置操作，走默认部分更新逻辑
    return super().partial_update(request, pk)

关键修改点：

事务包裹：用transaction.atomic()确保两个更新操作原子执行，避免中间状态触发约束。
分步更新：先排除当前项，将其他所有项设为False，再设置当前项为True，从逻辑上避免冲突。
修正序列化器：传入实例对象，确保是对现有数据的更新而非创建新数据。
避免重复操作：处理收藏设置时手动保存并返回响应，跳过super().partial_update的重复更新。

解决方案2：将逻辑下沉到模型层（更通用）

如果希望所有修改is_favorite的操作（包括后台管理、脚本）都自动维护约束，可以在模型的save方法中实现：

class QueryModel(models.Model):
    # ... 现有字段 ...

    def save(self, *args, **kwargs):
        if self.is_favorite:
            with transaction.atomic():
                # 将当前用户其他收藏项设为False
                QueryModel.objects.filter(user=self.user, is_favorite=True).exclude(pk=self.pk).update(is_favorite=False)
        super().save(*args, **kwargs)

这种方式无需修改视图逻辑，所有修改操作都会自动触发约束维护。

内容的提问来源于stack exchange，提问作者Pedro Daumas

Django中UniqueConstraint约束下更新收藏字段报错求助

相关代码 #

模型代码 #

视图集代码 #

URL配置代码 #

报错信息 #

问题原因与解决方案 #

问题根源 #

解决方案1：修改视图集的partial_update方法 #