Django bulk_create 如何批量创建关联对象

结论

跨数据库时，较稳妥的做法是分两个阶段批量写入：

1. 在应用层提前生成 Invoice 主键。
2. 批量插入所有 Invoice。
3. 根据已知的 invoice_id 构造全部 Transaction。
4. 批量插入所有 Transaction。
5. 将两次写入放在同一个 transaction.atomic() 事务中。

如果 Invoice 使用数据库生成的自增主键，调用 bulk_create() 后能否拿到每个对象的主键，取决于 Django 版本和数据库后端，因此不适合作为跨数据库方案。改用应用层生成的 UUID 主键后，可以在插入数据前确定外键关系。

推荐方案：预生成 UUID 主键

import uuid

from django.db import models


class Invoice(models.Model):
    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
    )
    balance = models.DecimalField(
        max_digits=6,
        decimal_places=2,
    )


class Transaction(models.Model):
    amount = models.DecimalField(
        max_digits=6,
        decimal_places=2,
    )
    invoice = models.ForeignKey(
        Invoice,
        on_delete=models.CASCADE,
        related_name="transactions",
        null=False,
    )

这里将反向关系名改成了 transactions。原来的 related_name="invoices" 不影响数据正确性，但使用 invoice.transactions.all() 更符合这段关系的实际含义。

计费时，先在内存中建立完整的对象关系：

from decimal import Decimal

from django.db import transaction


def run_billing(customers):
    invoices = []
    transactions = []

    for customer in customers:
        transaction_amounts = calculate_transaction_amounts(customer)
        balance = sum(transaction_amounts, start=Decimal("0.00"))

        invoice = Invoice(balance=balance)
        invoices.append(invoice)

        for amount in transaction_amounts:
            transactions.append(
                Transaction(
                    amount=amount,
                    invoice_id=invoice.pk,
                )
            )

    with transaction.atomic():
        Invoice.objects.bulk_create(
            invoices,
            batch_size=1000,
        )
        Transaction.objects.bulk_create(
            transactions,
            batch_size=5000,
        )

创建 Invoice 实例时，UUID 主键已经生成，即使对象还没有写入数据库，也可以提前设置 Transaction.invoice_id。实际写入顺序仍是先插入父记录，再插入子记录，因此不会违反外键约束。

SQL 执行次数主要取决于批次数量，而不是对象总数。例如，按照上面的批次设置写入 5,000 张发票和 100,000 条交易，通常只需执行少量批量插入，不需要进行 105,000 次单条插入。

batch_size 并非越大越好。不同数据库对 SQL 长度、绑定参数数量和单次事务资源有不同限制，需要结合具体数据库和数据规模调整。

不能修改主键时的处理方式

如果 Invoice 必须继续使用 AutoField 或 BigAutoField，可以增加一个由应用生成的唯一标识，并在批量插入后用它查询数据库主键。

import uuid

from django.db import models


class Invoice(models.Model):
    import_key = models.UUIDField(
        default=uuid.uuid4,
        unique=True,
        editable=False,
    )
    balance = models.DecimalField(
        max_digits=6,
        decimal_places=2,
    )

处理过程分为三步：

from decimal import Decimal

from django.db import transaction


def run_billing(customers):
    invoices = []
    pending_transactions = {}

    for customer in customers:
        amounts = calculate_transaction_amounts(customer)
        invoice = Invoice(
            balance=sum(amounts, start=Decimal("0.00")),
        )

        invoices.append(invoice)
        pending_transactions[invoice.import_key] = amounts

    with transaction.atomic():
        Invoice.objects.bulk_create(
            invoices,
            batch_size=1000,
        )

        saved_invoices = Invoice.objects.filter(
            import_key__in=pending_transactions.keys(),
        )

        invoice_ids = {
            invoice.import_key: invoice.pk
            for invoice in saved_invoices
        }

        transactions = []

        for import_key, amounts in pending_transactions.items():
            invoice_id = invoice_ids[import_key]

            for amount in amounts:
                transactions.append(
                    Transaction(
                        amount=amount,
                        invoice_id=invoice_id,
                    )
                )

        Transaction.objects.bulk_create(
            transactions,
            batch_size=5000,
        )

与使用 UUID 主键相比，这种方式多了一次查询，但整个过程仍然只有：

• 若干次 Invoice 批量插入；
• 若干次主键映射查询；
• 若干次 Transaction 批量插入。

import_key 必须设置数据库唯一约束。不要通过查询最近插入的记录或推算自增 ID 建立映射，因为并发任务和请求可能在此期间插入其他记录。

数据量较大时，import_key__in=... 查询也应分批执行，以免超过数据库的参数数量限制。

并发和事务边界

transaction.atomic() 可以保证父记录和子记录一起提交。如果交易插入失败，已经插入的发票也会回滚，其他连接通常不会看到缺少明细的半成品发票。

不过，事务无法阻止两个 Celery task 对同一客户重复计费。要避免重复执行，应依靠数据库约束，不能只在 Python 中先查询再判断。

一种常见做法是给发票增加计费周期或批次字段：

class Invoice(models.Model):
    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
    )
    customer = models.ForeignKey(
        "Customer",
        on_delete=models.PROTECT,
    )
    billing_period = models.DateField()
    balance = models.DecimalField(
        max_digits=6,
        decimal_places=2,
    )

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["customer", "billing_period"],
                name="unique_customer_billing_period",
            )
        ]

即使两个任务同时处理相同的客户和计费周期，数据库也只会接受一张对应的发票。应用需要捕获 IntegrityError，并将它作为重复任务处理。

如果计费过程还会读取并修改客户余额等共享状态，可以在事务中使用：

customer = (
    Customer.objects
    .select_for_update()
    .get(pk=customer_id)
)

一次锁定数千名客户可能会延长事务时间、增加等待，并提高死锁风险。需要锁定多行时，应按照固定顺序处理，同时尽量缩短锁的持有时间。

数据量很大时分块处理

将所有对象都保存在内存中，并放入一个超大事务，可以获得完整的全有或全无语义，但也会增加成本：

• Celery worker 会占用较多内存；
• 数据库事务会持续更长时间；
• 锁和旧版本数据的保留时间会增加；
• 一旦失败，整个计费周期都需要回滚。

可以按客户或计费批次分块处理：

from itertools import islice


def chunked(iterable, size):
    iterator = iter(iterable)

    while chunk := list(islice(iterator, size)):
        yield chunk


def run_billing(customers):
    for customer_batch in chunked(customers, 500):
        create_billing_batch(customer_batch)

每个批次内部仍采用两阶段批量插入：

from django.db import transaction


def create_billing_batch(customers):
    invoices, transactions = build_objects(customers)

    with transaction.atomic():
        Invoice.objects.bulk_create(
            invoices,
            batch_size=500,
        )
        Transaction.objects.bulk_create(
            transactions,
            batch_size=5000,
        )

分块可以减轻内存和长事务压力，但也意味着前面的批次可能已经提交，而后续批次仍可能失败。因此，需要保存明确的 billing_run_id 或计费周期，并通过唯一约束保证任务能够安全重试。

bulk_create() 的限制

使用前还需要确认这些行为是否符合业务要求：

• 不会调用每个对象的 save()。
• 不会触发常规的 pre_save 和 post_save 信号。
• 如果某些数据依赖 save() 或信号生成，需要提前在 Python 中计算，或者交给可靠的数据库默认值处理。
• 多对多关系不能随普通对象直接批量创建。需要先创建关系两端的对象，再批量写入中间表。
• bulk_create() 返回后是否会自动填充数据库生成的主键，取决于数据库后端，不能将其视为跨数据库保证。
• batch_size 只控制单批规模。Django 或数据库仍可能按照参数限制拆分语句。

在原生 Django 中，要实现跨数据库且并发安全的批量写入，重点不是寻找一条能同时插入父表和子表的特殊 API，而是提前获得稳定的父记录标识，再按照外键依赖顺序执行两组 bulk_create()。最直接的办法是使用 UUID 主键。如果必须保留自增主键，则应增加唯一关联键，并在插入父表后显式查询主键映射。

备注：内容仅供参考。