Django模型保存时限制单用户最多10条记录的最佳实践

你的思路方向是对的，但直接重写save方法存在并发写入的隐患——比如同一用户同时发起两个创建请求时，两次count()检查都可能返回9，导致最终插入11条记录。下面是几种更可靠的方案，按推荐优先级排序：

一、自定义管理器+原子事务（推荐） #

把记录数检查的逻辑封装到自定义管理器中，同时用原子事务+行锁避免并发问题，这种方式在Django层面实现，易维护且能覆盖大部分场景。

代码示例 #

from django.db import models, transaction
from django.contrib.auth.models import User

class MyModelManager(models.Manager):
    def create_for_user(self, user, **kwargs):
        # 原子事务确保检查、删除、插入是一个不可分割的操作
        with transaction.atomic():
            # 锁住该用户的所有MyModel记录，防止其他并发请求修改
            user_records = self.filter(user=user).select_for_update()
            if user_records.count() >= 10:
                # 删除最早创建的一条记录
                user_records.order_by('created_at').first().delete()
            # 创建新记录
            return self.create(user=user, **kwargs)

class MyModel(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    # 其他字段...

    objects = MyModelManager()

使用方式 #

# 创建新记录时统一用这个方法，避免直接调用create()
MyModel.objects.create_for_user(user=request.user, ...)

优势 #

• 逻辑集中，避免在多个地方重复写检查代码
• select_for_update会锁定用户的所有相关记录，彻底解决并发写入问题
• 仅在创建新记录时触发检查，更新现有记录不受影响

二、改进重写save方法 #

如果一定要用重写save的方式，必须加上原子事务和行锁，同时区分“创建新记录”和“更新旧记录”的场景：

from django.db import transaction

def save(self, *args, **kwargs):
    # 只有创建新记录时才执行检查逻辑
    if not self.pk:
        with transaction.atomic():
            user_records = MyModel.objects.filter(user=self.user).select_for_update()
            if user_records.count() >= 10:
                user_records.order_by('created_at').first().delete()
            super().save(*args, **kwargs)
    else:
        # 更新操作直接保存，不触发检查
        super().save(*args, **kwargs)

注意 #

• 这种方式的问题是，如果代码中使用bulk_create批量创建记录，save方法不会被调用，需要单独处理批量场景。

三、数据库触发器（底层约束） #

如果需要绝对的数据库层面约束（比如防止通过直接SQL插入数据绕过Django逻辑），可以用数据库触发器实现。但这种方式跨数据库兼容性差，调试成本高。

PostgreSQL触发器示例 #

1. 创建触发器函数：

CREATE OR REPLACE FUNCTION enforce_mymodel_user_limit()
RETURNS TRIGGER AS $$
BEGIN
    -- 检查当前用户的记录数是否达到10
    IF (SELECT COUNT(*) FROM myapp_mymodel WHERE user_id = NEW.user_id) >= 10 THEN
        -- 删除最早的一条记录
        DELETE FROM myapp_mymodel 
        WHERE user_id = NEW.user_id 
        ORDER BY created_at ASC 
        LIMIT 1;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

2. 创建触发器：

CREATE TRIGGER trigger_mymodel_limit
BEFORE INSERT ON myapp_mymodel
FOR EACH ROW EXECUTE FUNCTION enforce_mymodel_user_limit();

3. 在Django迁移中添加触发器（通过RunSQL）：

from django.db import migrations

class Migration(migrations.Migration):
    dependencies = [
        ('myapp', '0001_initial'),
    ]

    operations = [
        migrations.RunSQL("""
        CREATE OR REPLACE FUNCTION enforce_mymodel_user_limit()
        RETURNS TRIGGER AS $$
        BEGIN
            IF (SELECT COUNT(*) FROM myapp_mymodel WHERE user_id = NEW.user_id) >= 10 THEN
                DELETE FROM myapp_mymodel 
                WHERE user_id = NEW.user_id 
                ORDER BY created_at ASC 
                LIMIT 1;
            END IF;
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;

        CREATE TRIGGER trigger_mymodel_limit
        BEFORE INSERT ON myapp_mymodel
        FOR EACH ROW EXECUTE FUNCTION enforce_mymodel_user_limit();
        """),
    ]

优势与劣势 #

• 优势：不管通过ORM还是直接SQL操作，都能强制执行限制
• 劣势：跨数据库（比如MySQL、SQLite）需要写不同的触发器代码，调试困难，Django无法直接感知触发器逻辑

关键注意事项 #

• 并发问题：必须用transaction.atomic()+select_for_update()来保证检查、删除、插入的原子性，否则高并发场景下会出现超量记录。
• 批量操作：如果使用bulk_create，需要单独处理批量插入的逻辑，比如先计算需要删除的旧记录数量，再执行批量插入。

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