Todo tutorial de Django te enseña a montar un CRUD en media tarde. Modelo, serializer, viewset, router, y ya tienes una API funcionando. Lo que ningún tutorial te cuenta es qué pasa cuando esa API la usan cien personas a la vez, cuando dos peticiones intentan modificar la misma fila al mismo tiempo, o cuando un endpoint que "funciona bien" en local hace trescientas consultas a la base de datos en producción sin que nadie se dé cuenta hasta que la factura de infraestructura sube.
Esos problemas no aparecen en el tutorial porque no aparecen con poco tráfico ni con pocos datos. Aparecen cuando el proyecto ya está vivo, y para entonces ya cuestan más arreglarlos. Este artículo es un repaso de cuatro problemas concretos que me he encontrado trabajando con Django en producción: cómo diseñar bien las APIs, el problema de N+1 queries, las race conditions, y la idempotencia. No es teoría abstracta, cada sección tiene el código del problema y el código de la solución.
Diseñar APIs y endpoints que no te hagan pagar seis meses después
Django REST Framework te da varias formas de exponer un modelo como API, y la elegida en cada caso importa más de lo que parece al principio.
Para un CRUD estándar, un ModelViewSet con un router es la opción más rápida:
from rest_framework import viewsets, filters
from django_filters.rest_framework import DjangoFilterBackend
from .models import Paciente
from .serializers import PacienteSerializer
class PacienteViewSet(viewsets.ModelViewSet):
queryset = Paciente.objects.all()
serializer_class = PacienteSerializer
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
filterset_fields = ["activo", "clinica"]
search_fields = ["nombre", "apellidos", "email"]
ordering_fields = ["creado_en", "nombre"]Ojo con un detalle que se pasa por alto a menudo: filterset_fields, search_fields y ordering_fields no hacen nada por sí solos. Solo tienen efecto si declaras filter_backends con los backends correspondientes (o los configuras globalmente en REST_FRAMEWORK["DEFAULT_FILTER_BACKENDS"] en tu settings.py), y DjangoFilterBackend requiere tener instalado el paquete django-filter (pip install django-filter) y añadido django_filters a INSTALLED_APPS. Es un fallo silencioso: si te falta el filter_backends, el filtro simplemente no filtra, sin ningún error que te avise.
Con esto tienes listar, crear, leer, actualizar y borrar, más filtrado, búsqueda y ordenación. El problema aparece cuando el endpoint deja de ser un CRUD limpio: una acción que no encaja en las cinco operaciones estándar, una respuesta que combina datos de varios modelos, una regla de negocio que no es "guarda esto en la base de datos" sino "haz esto, y si falla, deshaz aquello otro". Ahí es donde forzar un ModelViewSet para todo empieza a generar código difícil de seguir, con lógica de negocio metida en el serializer o en un perform_create sobrecargado.
Mi regla es sencilla: ModelViewSet para lo que de verdad es un CRUD, y una APIView explícita para cualquier cosa que no lo sea:
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from django.db import transaction
class ConfirmarCitaView(APIView):
def post(self, request, cita_id):
with transaction.atomic():
cita = Cita.objects.select_for_update().get(id=cita_id)
if cita.estado != Cita.Estado.PENDIENTE:
return Response(
{"detail": "La cita ya no está pendiente de confirmación."},
status=status.HTTP_409_CONFLICT,
)
cita.estado = Cita.Estado.CONFIRMADA
cita.confirmada_en = timezone.now()
cita.save(update_fields=["estado", "confirmada_en"])
enviar_recordatorio.delay(cita.id)
return Response(PacienteCitaSerializer(cita).data)Este endpoint no es "actualizar un campo de un modelo", es "ejecutar una transición de estado con sus reglas y sus efectos secundarios". Forzarlo dentro de un update() de ModelViewSet habría significado sobrescribir ese método y acabar con un if gigante dentro de una función pensada para otra cosa.
Sobre versionado, si tu API la consumen clientes externos —una app móvil, un tercero—, versiona desde el primer día, aunque solo tengas una versión. Añadir /api/v1/ cuesta nada al principio y te ahorra una migración dolorosa el día que necesites cambiar la forma de una respuesta sin romper a los clientes que ya la consumen.
El problema de N+1: la consulta que no ves hasta que hay datos de verdad
Este es, probablemente, el problema de rendimiento más común en cualquier API construida con un ORM, y Django no es una excepción. Ocurre cuando, para renderizar una lista de N elementos, el código acaba haciendo una consulta adicional por cada elemento para traer sus datos relacionados, en vez de traerlos todos de una vez.
Imagina un serializer así:
class CitaSerializer(serializers.ModelSerializer):
paciente_nombre = serializers.CharField(source="paciente.nombre_completo")
profesional_nombre = serializers.CharField(source="profesional.nombre_completo")
class Meta:
model = Cita
fields = ["id", "fecha", "estado", "paciente_nombre", "profesional_nombre"]Y un viewset sencillo:
class CitaViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Cita.objects.all()
serializer_class = CitaSerializerEn local, con diez citas de prueba, esto funciona y parece perfectamente razonable. En producción, con quinientas citas en la respuesta, acabas de generar una consulta para traer las citas, más una consulta a paciente por cada cita, más otra a profesional por cada cita. Mil una consultas para una sola petición HTTP. Django Debug Toolbar te lo enseña sin piedad en cuanto lo instalas: una barra lateral con el número de queries de la página, y en un endpoint con N+1 esa cifra da vergüenza.
La solución para relaciones ForeignKey u OneToOne es select_related, que hace un único JOIN en SQL:
class CitaViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Cita.objects.select_related("paciente", "profesional")
serializer_class = CitaSerializerPara relaciones ManyToMany o inversas de ForeignKey, donde un JOIN único no basta porque el resultado sería una fila por cada combinación, la herramienta es prefetch_related, que lanza una segunda consulta separada y hace el emparejamiento en Python:
class ProfesionalViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Profesional.objects.prefetch_related("especialidades", "citas")
serializer_class = ProfesionalSerializerCon esto, listar profesionales con sus especialidades pasa de una consulta por profesional a exactamente dos consultas en total, sin importar cuántos profesionales haya.
Un caso más sutil es cuando necesitas un dato agregado, no solo la relación en sí. Si quieres mostrar cuántas citas tiene cada profesional, la tentación es calcularlo en Python iterando sobre profesional.citas.count() dentro del serializer, lo cual vuelve a caer en N+1. La forma correcta es dejar que la base de datos lo calcule con annotate:
from django.db.models import Count
queryset = Profesional.objects.annotate(
num_citas=Count("citas")
)Ahora num_citas es un campo más del queryset, calculado en una sola consulta SQL con un GROUP BY, y el serializer solo tiene que leerlo.
Mi recomendación práctica: instala django-debug-toolbar en desarrollo desde el primer endpoint que escribas, no cuando ya sospeches que algo va lento. Revisar el contador de queries de cada vista mientras la desarrollas es mucho más barato que hacerlo con una tabla de producción con cien mil filas y un cliente esperando al teléfono.
Race conditions: cuando dos peticiones llegan a la vez
Una race condition ocurre cuando el resultado final depende del orden exacto en el que se ejecutan dos operaciones concurrentes, y ese orden no está garantizado. En una API web, "dos peticiones a la vez" no es un caso raro: es Tuesday por la mañana con tráfico normal.
El ejemplo clásico es reservar el último hueco disponible de algo. Imagina un sistema de citas donde cada franja horaria solo admite un paciente:
def reservar_cita(profesional_id, fecha, hora, paciente):
existe = Cita.objects.filter(
profesional_id=profesional_id,
fecha=fecha,
hora=hora,
).exists()
if existe:
raise HuecoOcupadoError()
return Cita.objects.create(
profesional_id=profesional_id,
fecha=fecha,
hora=hora,
paciente=paciente,
)Este código es perfectamente correcto leído de arriba a abajo, y falla igualmente en producción. Si dos peticiones llegan casi al mismo tiempo, ambas pueden ejecutar el exists() antes de que ninguna haya hecho el create(). Las dos ven que el hueco está libre, las dos lo crean, y acabas con dos citas para el mismo profesional a la misma hora. El bug no está en la lógica, está en el hueco de tiempo entre "comprobar" y "actuar", que en inglés se suele llamar check-then-act.
Hay dos formas de arreglar esto, y no son excluyentes. Pero antes hay que entender por qué la solución "obvia" no funciona.
La tentación es meter un select_for_update() en la propia consulta que comprueba el hueco:
El problema es que select_for_update bloquea las filas que la consulta devuelve. Y en el caso que queremos proteger —la hora está libre, todavía no existe ninguna cita ahí—, la consulta devuelve cero filas. No se puede bloquear una fila que no existe, así que este select_for_update no bloquea absolutamente nada, y dos peticiones concurrentes pueden seguir colándose exactamente igual que antes. Es un error fácil de cometer porque el código parece correcto y hasta "suena" a que está usando el bloqueo bien.
Lo que sí funciona es bloquear una fila que sí existe y que ambas peticiones concurrentes van a tocar sin remedio: el propio profesional. Al bloquear esa fila, la segunda transacción tiene que esperar a que la primera termine antes de poder leer nada, y cuando por fin puede leer, la cita creada por la primera transacción ya es visible:
from django.db import transaction
def reservar_cita(profesional_id, fecha, hora, paciente):
with transaction.atomic():
profesional = Profesional.objects.select_for_update().get(id=profesional_id)
conflicto = Cita.objects.filter(
profesional=profesional, fecha=fecha, hora=hora,
).exists()
if conflicto:
raise HuecoOcupadoError()
return Cita.objects.create(
profesional=profesional, fecha=fecha, hora=hora, paciente=paciente,
)Esto funciona, pero con dos condiciones que hay que tener presentes. La primera es que protege exactamente el bloque de código que pasa por ahí: si otra parte de la aplicación crea citas sin pasar por esta misma función, el bloqueo no las ve y no las frena. La segunda es más de infraestructura: select_for_update() no tiene ningún efecto en SQLite, porque no soporta bloqueo de filas a nivel de base de datos — la propia documentación de Django lo dice explícitamente. Si pruebas esto en local con SQLite pensando que estás verificando el comportamiento real, no estás probando nada; necesitas Postgres o MySQL para ver el bloqueo en acción.
Por eso mi recomendación por defecto es la segunda vía: dejar que la propia base de datos garantice la regla, con una restricción de unicidad a nivel de tabla.
Con esto, aunque dos peticiones lleguen exactamente al mismo tiempo y ambas pasen la comprobación, la base de datos rechazará el segundo INSERT con un error de integridad. El código de la vista solo tiene que capturarlo y devolver una respuesta razonable:
from django.db import IntegrityError
def post(self, request):
try:
cita = reservar_cita(**datos)
except IntegrityError:
return Response(
{"detail": "Ese hueco ya no está disponible."},
status=status.HTTP_409_CONFLICT,
)
return Response(CitaSerializer(cita).data, status=status.HTTP_201_CREATED)La diferencia de fondo es esta: select_for_update protege un bloque de código concreto, mientras que una restricción de base de datos protege el dato en sí, sin importar desde dónde se intente escribir. En general, si la regla se puede expresar como una restricción de la base de datos, prefiero esa opción por defecto y reservo select_for_update para lógica de negocio más compleja que una constraint no puede capturar por sí sola —como calcular y descontar un stock disponible en el mismo movimiento.
Para ese caso, el de restar una cantidad de forma segura, otra herramienta útil es F(), que traslada la operación aritmética a la propia base de datos en vez de hacerla en Python:
from django.db.models import F
# Mal: lee el valor, resta en Python, y escribe. Vulnerable a race conditions.
producto = Producto.objects.get(id=producto_id)
producto.stock = producto.stock - cantidad
producto.save()
# Bien: la resta ocurre atómicamente en la base de datos.
Producto.objects.filter(id=producto_id).update(stock=F("stock") - cantidad)La versión con F() nunca lee el valor intermedio en Python, así que no importa cuántas peticiones concurrentes lo ejecuten: cada UPDATE se aplica de forma atómica sobre el valor que la base de datos tenga en ese instante. Eso sí, esta versión mínima no evita que el stock quede en negativo si cantidad es mayor que el disponible — es solo la pieza que resuelve la concurrencia. La versión completa, con la comprobación de disponibilidad incluida en la misma operación atómica, la ves un poco más abajo en el ejemplo de crear_pedido.
Idempotencia: cuando la misma petición llega más de una vez
Un endpoint es idempotente cuando ejecutarlo varias veces con los mismos datos produce el mismo resultado que ejecutarlo una sola vez. GET, PUT y DELETE son idempotentes por definición dentro de HTTP. POST no lo es, y ahí está el problema: un POST que crea un cargo, envía un email o genera un pedido, ejecutado dos veces, produce dos cargos, dos emails o dos pedidos.
Y un POST duplicado ocurre más de lo que parece: un usuario que hace doble clic en "pagar" porque el botón tardó en responder, una app móvil que reintenta automáticamente una petición porque la conexión se cortó justo después de que el servidor la procesara pero antes de que la respuesta llegara al cliente, o un webhook de un proveedor de pagos que reenvía el mismo evento porque no recibió confirmación a tiempo. En los tres casos, el cliente no tiene forma de saber si su primera petición llegó a completarse o no, así que lo razonable desde su punto de vista es reintentar. El error sería no plantearse la duplicidad como parte normal del diseño.
El patrón habitual es que el cliente genere una clave de idempotencia única por operación —normalmente un UUID— y la mande en una cabecera:
POST /api/v1/pagos/
Idempotency-Key: 8f14e45f-ceea-467e-bd9b-c7a01b6b1c1fEl servidor guarda esa clave asociada al resultado de la primera ejecución, y si vuelve a llegar la misma clave, devuelve el resultado guardado sin repetir la operación:
class IdempotencyKey(models.Model):
key = models.CharField(max_length=255, unique=True)
endpoint = models.CharField(max_length=255)
status_code = models.PositiveIntegerField(null=True, blank=True)
response_body = models.JSONField(null=True, blank=True)
creado_en = models.DateTimeField(auto_now_add=True)
class PagoView(APIView):
def post(self, request):
idempotency_key = request.headers.get("Idempotency-Key")
if not idempotency_key:
return Response(
{"detail": "Falta la cabecera Idempotency-Key."},
status=status.HTTP_400_BAD_REQUEST,
)
try:
registro = IdempotencyKey.objects.create(
key=idempotency_key,
endpoint=request.path,
)
except IntegrityError:
registro = IdempotencyKey.objects.get(key=idempotency_key)
if registro.response_body is None:
# La primera petición con esta clave todavía se está procesando
return Response(
{"detail": "Ya hay una solicitud en curso con esta clave."},
status=status.HTTP_409_CONFLICT,
)
return Response(registro.response_body, status=registro.status_code)
with transaction.atomic():
pago = procesar_pago(request.data)
response_data = PagoSerializer(pago).data
registro.status_code = status.HTTP_201_CREATED
registro.response_body = response_data
registro.save(update_fields=["status_code", "response_body"])
return Response(response_data, status=status.HTTP_201_CREATED)El orden importa, y es justo lo que cambia respecto a una primera versión ingenua de este patrón: la clave se registra antes de procesar el pago, no después. Si intentaras comprobar primero si la clave existe y solo procesar el pago si no existe, dejarías abierta la misma race condition que vimos en la sección anterior — dos peticiones con la misma clave podrían pasar la comprobación antes de que ninguna hubiera guardado nada, y las dos acabarían cobrando. Al hacer que el propio create() de IdempotencyKey actúe como el punto de sincronización —gracias a la restricción unique=True del campo key—, solo una petición puede "ganar" ese INSERT; la que pierde recibe un IntegrityError inmediato, antes de haber tocado nada relacionado con el pago.
También merece explicación el caso de registro.response_body is None: significa que otra petición ya reservó esa clave pero todavía no ha terminado de procesarse. Aquí no conviene reintentar el trabajo ni bloquear indefinidamente esperando; lo razonable es devolver un error que le diga al cliente que la petición original sigue en marcha. Es, de hecho, el mismo criterio que sigue la propia documentación de Stripe: si una petición llega con una clave que ya está en uso por otra que se está ejecutando en paralelo, no procesan un segundo intento.
Este patrón es exactamente el que usan Stripe y la mayoría de pasarelas de pago en su propia API, y si trabajas con webhooks de terceros, conviene aplicarlo también al recibirlos: guarda el identificador de evento que te manda el proveedor, y si ese evento ya lo procesaste, responde 200 OK sin repetir la acción. Un 200 en la segunda entrega, aunque no hagas nada, es preferible a que el proveedor interprete un error como que necesita reintentar de nuevo.
Atomicidad: que un fallo a medias no deje datos a medias
Ligado a todo lo anterior está el uso de transacciones. Cuando una operación implica varios pasos que deben ocurrir todos o ninguno —crear un pedido y descontar el stock, por ejemplo—, envolverlos en transaction.atomic() garantiza que si algo falla a mitad de camino, la base de datos vuelve exactamente al estado anterior, sin registros huérfanos ni stock descontado de un pedido que nunca se llegó a confirmar:
def crear_pedido(carrito, usuario):
with transaction.atomic():
pedido = Pedido.objects.create(usuario=usuario, total=carrito.total)
for item in carrito.items.all():
actualizado = Producto.objects.filter(
id=item.producto_id,
stock__gte=item.cantidad,
).update(stock=F("stock") - item.cantidad)
if not actualizado:
raise StockInsuficienteError(item.producto_id)
LineaPedido.objects.create(
pedido=pedido,
producto_id=item.producto_id,
cantidad=item.cantidad,
)
return pedidoAquí, si el stock de cualquier producto del carrito no alcanza, la excepción hace que Django deshaga toda la transacción: ni el pedido ni ninguna línea ni ningún descuento de stock quedan guardados. Y fíjate que la comprobación de stock (stock__gte=item.cantidad) va dentro del propio filter() del update(), no como una lectura previa en Python: así el chequeo y la resta ocurren en la misma operación atómica, sin el hueco de check-then-act que vimos antes con las citas.
Un detalle que se pasa por alto a menudo: si dentro de esa transacción necesitas disparar una tarea asíncrona —enviar un email de confirmación, encolar un job de Celery—, hacerlo con transaction.on_commit() en vez de directamente evita una clase entera de bugs difíciles de reproducir, donde la tarea se ejecuta contra una fila que, en el momento de leerla, la transacción que la creó todavía no se ha confirmado:
def crear_pedido(carrito, usuario):
with transaction.atomic():
pedido = Pedido.objects.create(usuario=usuario, total=carrito.total)
# ... resto de la lógica ...
transaction.on_commit(lambda: enviar_confirmacion.delay(pedido.id))
return pedidoon_commit garantiza que la tarea solo se dispara si la transacción se confirma con éxito, y solo después de que se haya confirmado, así que cuando el worker de Celery vaya a leer el pedido, este ya existe de forma consistente en la base de datos.
Conclusión: el código correcto no es el que funciona en local
Los cuatro problemas de este artículo comparten algo: todos son invisibles con poco tráfico y pocos datos, y todos se vuelven caros cuanto más tarde los detectas. Un N+1 no se nota con diez registros, una race condition no se nota si nunca pruebas con dos peticiones simultáneas de verdad, y la idempotencia no se echa en falta hasta el día que un cliente de pagos te reenvía un webhook duplicado y descubres que acabas de cobrar dos veces a la misma persona.
Ninguno de estos problemas se soluciona con más experiencia en abstracto, se solucionan con el hábito concreto de hacerte siempre las mismas preguntas al escribir un endpoint: ¿cuántas queries genera esto si la lista tiene mil elementos en vez de diez? ¿Qué pasa si esta función se ejecuta dos veces a la vez sobre la misma fila? ¿Qué pasa si esta petición llega duplicada? Si tienes respuesta a las tres antes de hacer merge, vas por buen camino. Si no la tienes, es buena señal de que toca pararse un momento antes de dar el endpoint por terminado.