Bulk operations en DRF: bulk_create, bulk_update y endpoints batch

Por que necesitas endpoints bulk

Una integracion B2B importa 5000 productos del catalogo una vez al dia. Si el cliente hace 5000 POST /api/productos/:

• Latencia: 5000 round-trips de ~50 ms = 4 minutos solo en latencia.
• 5000 transacciones SQL: cada INSERT con su BEGIN/COMMIT, fsync incluido.
• Rate limiting salta: tu UserRateThrottle de 1000/h tira 4000 peticiones a 429.
• Logs llenos de 5000 lineas iguales.

Un solo POST /api/productos/bulk/ con [{...}, {...}, ...] resuelve todo: una transaccion, una invocacion al ORM, una entrada de log.

Patron simple: serializer many=True

DRF lo soporta de forma nativa. Le pasas many=True y procesa una lista.

# productos/views.py
from rest_framework.viewsets import ModelViewSet
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework import status
from django.db import transaction

class ProductoViewSet(ModelViewSet):
    queryset = Producto.objects.all()
    serializer_class = ProductoSerializer

    @action(detail=False, methods=["post"], url_path="bulk")
    def bulk_create(self, request):
        if not isinstance(request.data, list):
            return Response({"error": "envia una lista"}, status=400)
        if len(request.data) > 1000:
            return Response({"error": "maximo 1000 items"}, status=413)

        serializer = self.get_serializer(data=request.data, many=True)
        serializer.is_valid(raise_exception=True)
        with transaction.atomic():
            serializer.save()
        return Response(serializer.data, status=status.HTTP_201_CREATED)

Funciona pero no es eficiente: por defecto serializer.save() con many invoca create() por cada item, lo que genera N inserts en lugar de uno solo.

Optimizacion real: ListSerializer.create con bulk_create

Override ListSerializer.create para usar Producto.objects.bulk_create(), que emite un unico INSERT con todos los valores.

# productos/serializers.py
from rest_framework import serializers
from productos.models import Producto

class ProductoListSerializer(serializers.ListSerializer):
    def create(self, validated_data):
        productos = [Producto(**item) for item in validated_data]
        return Producto.objects.bulk_create(productos, batch_size=500)

class ProductoSerializer(serializers.ModelSerializer):
    class Meta:
        model        = Producto
        fields       = ["id", "nombre", "precio", "stock", "categoria"]
        list_serializer_class = ProductoListSerializer

batch_size=500 divide la insercion en lotes de 500 filas para no enviar a Postgres un INSERT de 5 MB de una vez.

bulk_create no llama a save() de cada modelo, lo que tambien significa que no dispara senales pre_save/post_save. Si dependes de senales (auditoria, cache invalidation), invocalas manualmente o usa el patron menos optimo de un loop con save() dentro de transaction.atomic().

Bulk update

class ProductoListSerializer(serializers.ListSerializer):
    def update(self, instances, validated_data):
        # Mapeamos id -> nuevos valores
        mapa = {item["id"]: item for item in validated_data}
        for instance in instances:
            for k, v in mapa.get(instance.id, {}).items():
                setattr(instance, k, v)
        Producto.objects.bulk_update(
            instances, fields=["precio", "stock"], batch_size=500
        )
        return instances

Action en el ViewSet:

@action(detail=False, methods=["put"], url_path="bulk")
def bulk_update(self, request):
    ids = [it["id"] for it in request.data]
    instances = list(Producto.objects.filter(id__in=ids))
    serializer = self.get_serializer(instances, data=request.data, many=True, partial=True)
    serializer.is_valid(raise_exception=True)
    with transaction.atomic():
        serializer.save()
    return Response(serializer.data)

Bulk delete

@action(detail=False, methods=["delete"], url_path="bulk")
def bulk_delete(self, request):
    ids = request.data.get("ids", [])
    if not ids or len(ids) > 1000:
        return Response({"error": "ids vacio o > 1000"}, status=400)
    deleted, _ = Producto.objects.filter(id__in=ids).delete()
    return Response({"deleted": deleted})

Decision clave: todo o nada frente a partial success

Todo o nada (transaction.atomic)

POST /api/productos/bulk/
[ {ok}, {ok}, {malo}, {ok} ]

Respuesta: 400 Bad Request
{ "errors": [null, null, {"precio": ["debe ser positivo"]}, null] }
Resultado en BBDD: nada creado

Util para imports atomicos donde la integridad importa. Sencillo de entender.

Partial success (multi-status 207)

HTTP/1.1 207 Multi-Status
[
  { "status": 201, "id": 1, "data": {...} },
  { "status": 201, "id": 2, "data": {...} },
  { "status": 422, "error": {"precio": ["debe ser positivo"]} },
  { "status": 201, "id": 3, "data": {...} }
]

3 productos creados, 1 fallo. Util cuando el cliente necesita procesar lo que pueda y reintentar solo los fallos.

@action(detail=False, methods=["post"], url_path="bulk-partial")
def bulk_create_partial(self, request):
    resultados = []
    for item in request.data:
        s = self.get_serializer(data=item)
        if s.is_valid():
            s.save()
            resultados.append({"status": 201, "data": s.data})
        else:
            resultados.append({"status": 422, "error": s.errors})
    # 207 si hubo mezcla, 201 si todo bien, 422 si todo mal
    if all(r["status"] == 201 for r in resultados):
        code = 201
    elif all(r["status"] != 201 for r in resultados):
        code = 422
    else:
        code = 207
    return Response(resultados, status=code)

Decision: transaction.atomic por defecto, partial success solo si el dominio lo justifica (imports muy grandes donde reintentar todo es muy caro).

Limites y proteccion

MAX_BULK = 1000

@action(detail=False, methods=["post"], url_path="bulk")
def bulk_create(self, request):
    if not isinstance(request.data, list):
        return Response({"error": "lista requerida"}, status=400)
    if len(request.data) > MAX_BULK:
        return Response(
            {"error": f"max {MAX_BULK} items por bulk; usa multiples llamadas"},
            status=413,  # Payload Too Large
        )
    ...

Sin limite, un cliente puede enviar 10 millones de objetos y agotar la RAM del worker. 413 es la respuesta correcta segun RFC 9110.

Diagrama de comparacion

flowchart TB
    subgraph Sin_bulk
        A1[POST /productos/]
        A2[POST /productos/]
        A3[POST /productos/]
        AN[POST /productos/]
    end
    subgraph Con_bulk
        B1[POST /productos/bulk/<br>1 peticion, lista N items]
    end
    A1 --> SQL1[INSERT]
    A2 --> SQL2[INSERT]
    A3 --> SQL3[INSERT]
    AN --> SQLN[INSERT]
    B1 --> SQLBULK[INSERT...VALUES batch_size=500]
    SQL1 -.-> RES1[5000 viajes RTT]
    SQLBULK -.-> RES2[1 viaje RTT, 10 INSERTs grandes]

Documentar bulk en OpenAPI

drf-spectacular puede documentar listas:

from drf_spectacular.utils import extend_schema, OpenApiTypes

@extend_schema(
    request   = ProductoSerializer(many=True),
    responses = {201: ProductoSerializer(many=True), 207: OpenApiTypes.OBJECT},
    description = "Crea hasta 1000 productos en una sola peticion."
)
@action(detail=False, methods=["post"], url_path="bulk")
def bulk_create(self, request):
    ...

Buenas practicas

• Indica al cliente el tamano maximo en la documentacion y devuelve 413 al pasarse.
• Usa batch_size en bulk_create y bulk_update para evitar INSERTs gigantes.
• No pretendas reemplazar los endpoints unitarios. Bulk es complementario, no sustituto.
• Aumenta el throttling para endpoints bulk si tiene sentido (scoped con menos req/h pero mas items por req).
• Considera que bulk_create no genera senales y, si lo necesitas, dispara los efectos manualmente despues.