Django 3.2+PostgreSQL：如何为MediaFile添加唯一使用约束？

一、数据库层面：PostgreSQL触发器（强约束，推荐） #

Django内置约束无法跨表统计关联次数，因此我们可以利用PostgreSQL的触发器实现数据库级别的强制约束，确保每个MediaFile被引用次数不超过1。

步骤1：创建触发器函数 #

编写PL/pgSQL函数，用于检查MediaFile的引用计数：

CREATE OR REPLACE FUNCTION check_mediafile_usage()
RETURNS TRIGGER AS $$
BEGIN
    -- 校验新的profile_picture是否已被使用
    IF NEW.profile_picture_id IS NOT NULL THEN
        IF (SELECT COUNT(*) FROM customer
            WHERE profile_picture_id = NEW.profile_picture_id OR id_picture_id = NEW.profile_picture_id) >= 1 THEN
            RAISE EXCEPTION 'MediaFile % is already in use', NEW.profile_picture_id;
        END IF;
    END IF;

    -- 校验新的id_picture是否已被使用
    IF NEW.id_picture_id IS NOT NULL THEN
        IF (SELECT COUNT(*) FROM customer
            WHERE profile_picture_id = NEW.id_picture_id OR id_picture_id = NEW.id_picture_id) >= 1 THEN
            RAISE EXCEPTION 'MediaFile % is already in use', NEW.id_picture_id;
        END IF;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

步骤2：创建触发器 #

给customer表添加INSERT和UPDATE触发器，触发上述函数：

CREATE TRIGGER trigger_check_mediafile_usage
BEFORE INSERT OR UPDATE OF profile_picture_id, id_picture_id
ON customer
FOR EACH ROW
EXECUTE FUNCTION check_mediafile_usage();

步骤3：在Django迁移中执行SQL #

创建空迁移文件，通过RunSQL执行上述SQL：

from django.db import migrations, models

class Migration(migrations.Migration):
    dependencies = [
        # 替换为你的app名和上一个迁移编号，例如('myapp', '0001_initial')
        ('myapp', '0002_xxxx'),
    ]

    operations = [
        migrations.RunSQL("""
            CREATE OR REPLACE FUNCTION check_mediafile_usage()
            RETURNS TRIGGER AS $$
            BEGIN
                IF NEW.profile_picture_id IS NOT NULL THEN
                    IF (SELECT COUNT(*) FROM customer
                        WHERE profile_picture_id = NEW.profile_picture_id OR id_picture_id = NEW.profile_picture_id) >= 1 THEN
                        RAISE EXCEPTION 'MediaFile % is already in use', NEW.profile_picture_id;
                    END IF;
                END IF;

                IF NEW.id_picture_id IS NOT NULL THEN
                    IF (SELECT COUNT(*) FROM customer
                        WHERE profile_picture_id = NEW.id_picture_id OR id_picture_id = NEW.id_picture_id) >= 1 THEN
                        RAISE EXCEPTION 'MediaFile % is already in use', NEW.id_picture_id;
                    END IF;
                END IF;

                RETURN NEW;
            END;
            $$ LANGUAGE plpgsql;
        """),
        migrations.RunSQL("""
            CREATE TRIGGER trigger_check_mediafile_usage
            BEFORE INSERT OR UPDATE OF profile_picture_id, id_picture_id
            ON customer
            FOR EACH ROW
            EXECUTE FUNCTION check_mediafile_usage();
        """),
    ]

二、Django层面：模型校验（补充拦截） #

数据库触发器是强约束，但可以在应用层增加校验，提前拦截错误提升体验：

方法1：重写Customer的clean方法 #

from django.core.exceptions import ValidationError
from django.db.models import Q

class Customer(models.Model):
    profile_picture = models.ForeignKey('MediaFile', on_delete=models.PROTECT, related_name='profile_uses')
    id_picture = models.ForeignKey('MediaFile', on_delete=models.PROTECT, related_name='id_uses')

    def clean(self):
        super().clean()
        # 校验自身两个字段是否引用同一文件
        if self.profile_picture == self.id_picture:
            raise ValidationError("头像和证件照不能是同一个文件")
        
        # 校验profile_picture是否已被使用
        if self.profile_picture:
            usage_count = Customer.objects.filter(
                Q(profile_picture=self.profile_picture) | Q(id_picture=self.profile_picture)
            ).exclude(pk=self.pk).count()
            if usage_count >= 1:
                raise ValidationError(f"文件ID {self.profile_picture.id}已被使用")
        
        # 校验id_picture是否已被使用
        if self.id_picture:
            usage_count = Customer.objects.filter(
                Q(profile_picture=self.id_picture) | Q(id_picture=self.id_picture)
            ).exclude(pk=self.pk).count()
            if usage_count >= 1:
                raise ValidationError(f"文件ID {self.id_picture.id}已被使用")

方法2：使用pre_save信号 #

若需统一校验逻辑，可通过信号实现：

from django.db.models.signals import pre_save
from django.dispatch import receiver
from django.core.exceptions import ValidationError
from django.db.models import Q

@receiver(pre_save, sender=Customer)
def check_mediafile_usage(sender, instance, **kwargs):
    if instance.profile_picture == instance.id_picture:
        raise ValidationError("头像和证件照不能是同一个文件")
    
    if instance.profile_picture:
        usage_count = Customer.objects.filter(
            Q(profile_picture=instance.profile_picture) | Q(id_picture=instance.profile_picture)
        ).exclude(pk=instance.pk).count()
        if usage_count >= 1:
            raise ValidationError(f"文件ID {instance.profile_picture.id}已被使用")
    
    if instance.id_picture:
        usage_count = Customer.objects.filter(
            Q(profile_picture=instance.id_picture) | Q(id_picture=instance.id_picture)
        ).exclude(pk=instance.pk).count()
        if usage_count >= 1:
            raise ValidationError(f"文件ID {instance.id_picture.id}已被使用")

原方案失效原因 #

Django的CheckConstraint仅能基于当前表的物理字段做约束，无法引用关联表的动态统计数据（例如number_of_usages这类注解字段是查询时生成的临时数据，并非数据库表的固定字段），因此无法实现跨表的引用次数限制。

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