Managers personalizados y consultas avanzadas

Managers personalizados

El manager por defecto de Django es objects, que proporciona la QuerySet API estándar. Es posible crear managers personalizados para encapsular consultas complejas y reutilizables en un único lugar:

from django.db import models
from django.utils import timezone

class ProductoActivoManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(activo=True)

class ProductoManager(models.Manager):
    def en_oferta(self):
        return self.filter(precio_oferta__isnull=False, activo=True)

    def recientes(self, dias=30):
        fecha_limite = timezone.now() - timezone.timedelta(days=dias)
        return self.filter(fecha_creacion__gte=fecha_limite)

    def por_categoria(self, categoria):
        return self.filter(categoria__slug=categoria)

class Producto(models.Model):
    nombre = models.CharField(max_length=100)
    precio = models.DecimalField(max_digits=10, decimal_places=2)
    precio_oferta = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
    activo = models.BooleanField(default=True)
    fecha_creacion = models.DateTimeField(auto_now_add=True)
    categoria = models.ForeignKey('Categoria', on_delete=models.CASCADE)

    objects = ProductoManager()           # Manager principal
    activos = ProductoActivoManager()     # Manager alternativo

    class Meta:
        ordering = ['-fecha_creacion']

Uso de los managers personalizados:

# Productos en oferta
Producto.objects.en_oferta()

# Productos recientes de los últimos 7 días
Producto.objects.recientes(dias=7)

# Solo productos activos (usando el manager alternativo)
Producto.activos.all()

Objetos Q: consultas con OR y AND complejos

Los objetos Q permiten construir condiciones lógicas complejas que el ORM estándar no puede expresar con simples filter() encadenados:

from django.db.models import Q

# OR: productos baratos O en oferta
productos = Producto.objects.filter(
    Q(precio__lt=20) | Q(precio_oferta__isnull=False)
)

# AND con OR anidado: activos Y (baratos O recientes)
productos = Producto.objects.filter(
    Q(activo=True) & (Q(precio__lt=50) | Q(fecha_creacion__year=2025))
)

# NOT: productos que NO están agotados
productos = Producto.objects.filter(~Q(stock=0))

# Combinando con kwargs normales
productos = Producto.objects.filter(
    Q(nombre__icontains='django') | Q(descripcion__icontains='django'),
    activo=True,  # AND implícito con los Q anteriores
)

Objetos F: operaciones entre campos

Los objetos F referencian campos del modelo para operar sobre ellos directamente en la base de datos, evitando cargar los datos en Python:

from django.db.models import F

# Incrementar el stock de todos los productos en 10 unidades (sin traer datos a Python)
Producto.objects.update(stock=F('stock') + 10)

# Aplicar un descuento del 10% a productos caros
Producto.objects.filter(precio__gte=100).update(
    precio_oferta=F('precio') * 0.9
)

# Filtrar productos donde el precio de oferta es menor al precio original
Producto.objects.filter(precio_oferta__lt=F('precio'))

# Comparar dos campos del mismo modelo
from django.db.models import ExpressionWrapper, DecimalField
descuento = ExpressionWrapper(
    F('precio') - F('precio_oferta'),
    output_field=DecimalField()
)
Producto.objects.annotate(ahorro=descuento).filter(ahorro__gt=10)

SQL raw

Cuando el ORM no puede expresar una consulta compleja, es posible ejecutar SQL directamente:

# Usando raw() - devuelve objetos del modelo
productos = Producto.objects.raw(
    'SELECT * FROM catalogo_producto WHERE precio < %s ORDER BY nombre',
    [50]
)
for p in productos:
    print(p.nombre, p.precio)

# Usando connection.cursor() para consultas que no devuelven modelos
from django.db import connection

def obtener_estadisticas():
    with connection.cursor() as cursor:
        cursor.execute('''
            SELECT categoria_id, COUNT(*) as total, AVG(precio) as precio_medio
            FROM catalogo_producto
            WHERE activo = TRUE
            GROUP BY categoria_id
        ''')
        columnas = [col[0] for col in cursor.description]
        return [dict(zip(columnas, fila)) for fila in cursor.fetchall()]

Transacciones con atomic()

El decorador @transaction.atomic y el gestor de contexto with transaction.atomic() garantizan que un bloque de operaciones se ejecuta como una transacción única:

from django.db import transaction

@transaction.atomic
def transferir_stock(producto_origen_id, producto_destino_id, cantidad):
    """Transfiere stock entre productos de forma atómica."""
    origen = Producto.objects.select_for_update().get(id=producto_origen_id)
    destino = Producto.objects.select_for_update().get(id=producto_destino_id)

    if origen.stock < cantidad:
        raise ValueError(f'Stock insuficiente: {origen.stock} disponibles')

    origen.stock = F('stock') - cantidad
    destino.stock = F('stock') + cantidad
    origen.save(update_fields=['stock'])
    destino.save(update_fields=['stock'])

# Como gestor de contexto
def procesar_pedido(pedido_id):
    with transaction.atomic():
        pedido = Pedido.objects.get(id=pedido_id)
        pedido.estado = 'procesando'
        pedido.save()
        for linea in pedido.lineas.all():
            linea.producto.stock = F('stock') - linea.cantidad
            linea.producto.save(update_fields=['stock'])
        pedido.estado = 'completado'
        pedido.save()

select_related y prefetch_related

El problema N+1 ocurre cuando Django ejecuta una consulta adicional por cada objeto de un QuerySet al acceder a sus relaciones. select_related y prefetch_related lo resuelven:

# PROBLEMA: N+1 consultas (1 para productos + 1 por cada producto para obtener su categoría)
productos = Producto.objects.all()
for p in productos:
    print(p.categoria.nombre)  # Cada acceso genera una nueva consulta SQL

# SOLUCIÓN con select_related (JOIN SQL): óptimo para ForeignKey y OneToOneField
productos = Producto.objects.select_related('categoria', 'proveedor').all()
for p in productos:
    print(p.categoria.nombre)  # Sin consultas adicionales

# SOLUCIÓN con prefetch_related: óptimo para ManyToManyField y relaciones inversas
pedidos = Pedido.objects.prefetch_related('lineas__producto').all()
for pedido in pedidos:
    for linea in pedido.lineas.all():  # Sin consultas adicionales
        print(linea.producto.nombre)

# Combinando ambos
Pedido.objects.select_related('cliente').prefetch_related(
    'lineas__producto__categoria'
).filter(estado='pendiente')

Aggregations y annotations

from django.db.models import Count, Avg, Max, Min, Sum

# Estadísticas globales
Producto.objects.aggregate(
    total=Count('id'),
    precio_medio=Avg('precio'),
    precio_max=Max('precio'),
    precio_min=Min('precio'),
    valor_total_stock=Sum(F('precio') * F('stock'))
)

# Anotaciones por objeto (agrega un campo calculado a cada resultado)
from django.db.models import Count
categorias = Categoria.objects.annotate(
    num_productos=Count('producto'),
    precio_medio=Avg('producto__precio')
).filter(num_productos__gt=0).order_by('-num_productos')

Los managers personalizados, los objetos Q y F, y las optimizaciones con select_related y prefetch_related son herramientas indispensables para construir aplicaciones Django de alto rendimiento con código limpio y mantenible.