PAYATHON 2026

Django 如何创建每月定期付款并写入 Credit 表

支付老李

结论

建议把“生成月度付款记录”写成可重复执行的幂等业务函数，再由定时任务每月调用。生产环境一般使用 Celery 和 django-celery-beat；项目规模较小时，也可以使用 Django management command 配合系统 cron。

不要把定时逻辑放进视图、save() 方法或 Django 启动代码，否则可能因重复请求、多进程部署或服务重启而多次生成付款记录。数据库中也要增加唯一约束，保证同一救助对象每个月最多只有一条付款记录。

以下示例基于两个假设：

  • “2500k”和“2000k”指的是 2500.00 和 2000.00。如果 k 表示“千”，金额应改为 2500000.00 和 2000000.00。
  • 年龄恰好为 50 岁时发放 2000.00。如果实际规则不同，调整判断条件即可。

调整 Credit 模型

DecimalField 必须设置 decimal_places。建议同时保存付款月份，并为“收款人 + 付款月份”添加唯一约束。

from django.db import models
from django.utils.translation import gettext_lazy as _


class Destitute(models.Model):
    GENDER = (("male", "male"), ("female", "female"))

    name = models.CharField(max_length=100)
    acc_no = models.CharField(_("Bank Account"), max_length=50)
    age = models.PositiveIntegerField(default=1)
    sex = models.CharField(_("Gender"), max_length=64, choices=GENDER)
    created_on = models.DateTimeField(_("Date Joined"), auto_now_add=True)

    def __str__(self):
        return self.name


class Credit(models.Model):
    credit_title = models.CharField(_("Payment Title"), max_length=50)
    payee = models.ForeignKey(
        Destitute,
        related_name="destitute_paid",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    total_amount = models.DecimalField(
        max_digits=12,
        decimal_places=2,
    )
    payment_date = models.DateField()
    payment_month = models.DateField(
        help_text="使用当月第一天表示付款月份",
    )

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["payee", "payment_month"],
                name="unique_credit_per_payee_month",
            ),
        ]

    def __str__(self):
        return self.credit_title

修改模型后执行迁移：

python manage.py makemigrations
python manage.py migrate

如果数据库中已经有 Credit 数据，新增非空字段时需要先设置默认值。也可以分两次迁移：第一次允许 null=True，补齐历史数据后，再将字段改为非空。

编写月度付款函数

把业务逻辑放在普通服务函数中，例如 payments/services.py：

from datetime import date
from decimal import Decimal

from django.db import transaction

from .models import Credit, Destitute


AMOUNT_OVER_50 = Decimal("2500.00")
AMOUNT_50_OR_UNDER = Decimal("2000.00")


@transaction.atomic
def generate_monthly_credits(run_date=None):
    run_date = run_date or date.today()
    payment_month = run_date.replace(day=1)

    created_count = 0

    for destitute in Destitute.objects.iterator():
        amount = (
            AMOUNT_OVER_50
            if destitute.age > 50
            else AMOUNT_50_OR_UNDER
        )

        _, created = Credit.objects.get_or_create(
            payee=destitute,
            payment_month=payment_month,
            defaults={
                "credit_title": payment_month.strftime(
                    "Monthly Payment %Y-%m"
                ),
                "total_amount": amount,
                "payment_date": run_date,
            },
        )

        if created:
            created_count += 1

    return created_count

这段代码通过 get_or_create() 和数据库唯一约束保证幂等性。即使任务因重试、误操作或多个 worker 同时运行而执行多次，同一个人也不会收到重复的当月付款记录。

iterator() 可以防止程序一次性把整张 Destitute 表加载到内存中。如果数据量很大，还可以使用批量查询和 bulk_create(ignore_conflicts=True)。对于普通规模的数据，get_or_create() 更容易理解，也更容易保证结果正确。

使用 Celery 定期执行

安装 Celery，并按照部署环境选择消息代理，通常可以使用 Redis 或 RabbitMQ：

pip install celery django-celery-beat

在 settings.py 中启用应用：

INSTALLED_APPS = [
    # ...
    "django_celery_beat",
]

接着迁移调度表：

python manage.py migrate django_celery_beat

定义 Celery 任务：

# payments/tasks.py

from celery import shared_task

from .services import generate_monthly_credits


@shared_task
def create_monthly_credit_records():
    return generate_monthly_credits()

可以在 settings.py 中配置每月执行的任务：

from celery.schedules import crontab


CELERY_BEAT_SCHEDULE = {
    "create-monthly-credit-records": {
        "task": "payments.tasks.create_monthly_credit_records",
        "schedule": crontab(
            minute=0,
            hour=2,
            day_of_month="1",
        ),
    },
}

这项配置会在每月 1 日凌晨 2 点执行任务。部署时还要启动 Celery worker 和 beat：

celery -A config worker --loglevel=INFO
celery -A config beat --loglevel=INFO

将 config 替换为包含项目 settings.py 的 Django 项目包名。如果需要通过 Django Admin 动态调整执行时间，可以在后台使用 django-celery-beat 的 PeriodicTask 配置，无需把调度时间写死在代码中。

小型项目的简单方案

如果不想引入 Celery，可以创建 management command，再由服务器上的 cron 每月运行。

# payments/management/commands/generate_monthly_credits.py

from django.core.management.base import BaseCommand

from payments.services import generate_monthly_credits


class Command(BaseCommand):
    help = "Generate monthly credit records"

    def handle(self, *args, **options):
        created_count = generate_monthly_credits()

        self.stdout.write(
            self.style.SUCCESS(
                f"Created {created_count} monthly credit records."
            )
        )

先手动验证命令：

python manage.py generate_monthly_credits

Linux cron 示例：

0 2 1 * * /path/to/venv/bin/python /path/to/project/manage.py generate_monthly_credits

cron 比较轻量，但没有 Celery 提供的任务重试、集中监控和数据库调度功能。如果项目已经使用 Redis、RabbitMQ 或其他后台任务，建议选择 Celery。若系统中只有一个简单的月度任务，management command 配合 cron 通常就够了。

注意事项

  • 先确认年龄恰好为 50 岁时采用哪档付款标准。
  • 金额要使用 Decimal，不要使用 float。
  • 如果年龄会随时间变化，长期保存 age 并不可靠。更合适的做法是保存出生日期，例如 date_of_birth，在执行付款时计算当月年龄。
  • payment_month 应统一保存为当月第一天，防止同一个月出现多个不同日期。
  • 定时任务使用的时区应与 TIME_ZONE、Celery 时区和服务器时区一致。
  • 银行提交和数据库记账最好使用不同状态，例如 pending、submitted、paid、failed。创建 Credit 记录不代表银行付款已经成功。
  • 如果删除救助对象后仍要保留付款审计记录，使用 SET_NULL 是合理的。不过，不同数据库对唯一约束中 NULL 的处理可能不同，因此历史记录最好不要继续参与新付款的生成逻辑。
  • 上线前需要测试重复执行、任务中断后的重试、空数据表、年龄边界和金额上限等情况。

备注：内容仅供参考。