Django中如何根据用户订阅套餐限制生产者内容上传数量

1. 基础模型定义 #

首先需要梳理套餐、生产者账号、上传内容三类数据的关联关系，示例模型代码如下：

from django.db import models
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.utils import timezone

User = get_user_model()

# 订阅套餐表，存储不同套餐的上传限额
class SubscriptionPlan(models.Model):
    name = models.CharField(max_length=50, unique=True)  # 例：Free/Pro/Enterprise
    max_upload_limit = models.PositiveIntegerField(default=1)  # Pro套餐此处设置为5即可匹配需求
    price = models.DecimalField(max_digits=6, decimal_places=2, default=0)

    def __str__(self):
        return self.name

# 生产者账号表，关联当前生效的订阅
class Producer(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="producer_profile")
    active_subscription = models.ForeignKey(SubscriptionPlan, on_delete=models.SET_NULL, null=True, blank=True)
    subscription_valid_until = models.DateTimeField(null=True, blank=True)  # 订阅有效期，用于判断订阅是否生效

    def __str__(self):
        return f"{self.user.username} - {self.active_subscription.name if self.active_subscription else '无有效订阅'}"

# 上传内容表，关联所属生产者
class UploadedContent(models.Model):
    producer = models.ForeignKey(Producer, on_delete=models.CASCADE, related_name="uploaded_contents")
    title = models.CharField(max_length=200)
    content_file = models.FileField(upload_to="producer_contents/")
    upload_time = models.DateTimeField(auto_now_add=True)
    is_deleted = models.BooleanField(default=False)  # 软删除标记，统计额度时可过滤已删除内容

2. 上传限制校验实现 #

根据项目技术栈选择对应校验方式，建议至少保留模型层校验，避免规则绕过：

方式1：模型层全局校验（推荐） #

重写上传内容模型的clean方法，所有场景下保存数据都会触发校验：

class UploadedContent(models.Model):
    # 上述字段定义不变
    def clean(self):
        super().clean()
        # 校验订阅有效性
        if not self.producer.active_subscription or self.producer.subscription_valid_until < timezone.now():
            raise ValidationError("当前无有效订阅，无法上传内容")
        # 统计已上传有效内容数量
        uploaded_count = self.producer.uploaded_contents.filter(is_deleted=False).count()
        max_limit = self.producer.active_subscription.max_upload_limit
        # 校验额度
        if uploaded_count >= max_limit:
            raise ValidationError(f"当前{self.producer.active_subscription.name}套餐最多仅可上传{max_limit}份内容，请升级套餐后重试")

如果是自定义接口/视图上传，需要手动调用full_clean()触发校验，使用Django原生ModelForm、Admin后台则会自动触发。

方式2：DRF 接口层校验 #

如果使用DRF开发接口，可在序列化器中添加校验逻辑：

from rest_framework import serializers
from .models import UploadedContent

class UploadContentSerializer(serializers.ModelSerializer):
    class Meta:
        model = UploadedContent
        fields = ["id", "title", "content_file", "upload_time"]
        read_only_fields = ["producer", "upload_time"]

    def validate(self, attrs):
        producer = self.context["request"].user.producer_profile
        # 订阅有效性校验
        if not producer.active_subscription or producer.subscription_valid_until < timezone.now():
            raise serializers.ValidationError("无有效订阅，无法上传内容")
        # 额度校验
        uploaded_count = producer.uploaded_contents.filter(is_deleted=False).count()
        max_limit = producer.active_subscription.max_upload_limit
        if uploaded_count >= max_limit:
            raise serializers.ValidationError(f"当前套餐最多可上传{max_limit}份内容，已达上限")
        attrs["producer"] = producer
        return attrs

方式3：传统Django视图校验 #

如果使用原生函数视图开发，可在视图逻辑中提前校验：

from django.shortcuts import render, redirect
from django.contrib import messages
from .forms import UploadContentForm

def upload_content(request):
    if request.method == "POST":
        form = UploadContentForm(request.POST, request.FILES)
        producer = request.user.producer_profile
        # 提前校验额度，避免文件上传后才拦截
        if not producer.active_subscription or producer.subscription_valid_until < timezone.now():
            messages.error(request, "无有效订阅，无法上传内容")
            return redirect("upload_page")
        uploaded_count = producer.uploaded_contents.filter(is_deleted=False).count()
        if uploaded_count >= producer.active_subscription.max_upload_limit:
            messages.error(request, "已达当前套餐上传上限，请升级套餐后重试")
            return redirect("upload_page")
        if form.is_valid():
            content = form.save(commit=False)
            content.producer = producer
            content.save()
            messages.success(request, "上传成功")
            return redirect("content_list")
    else:
        form = UploadContentForm()
    return render(request, "upload.html", {"form": form})

3. 补充优化建议 #

• 可新增单独接口返回当前用户剩余上传额度，前端上传前提前校验，优化用户体验（前端校验仅做体验优化，不能替代后端校验）
• 如果套餐支持升级/降级，无需修改校验逻辑，直接修改SubscriptionPlan对应套餐的max_upload_limit即可自动生效
• 统计上传数量时记得过滤软删除、审核未通过等无效内容，避免额度计算错误

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