heapq en Python: colas de prioridad

Cuando una aplicación debe procesar siempre el elemento más urgente, pequeño o cercano, ordenar toda la colección después de cada cambio suele ser innecesario. Los planificadores, simuladores, colas de tareas, algoritmos de rutas y análisis de grandes conjuntos de datos necesitan acceder rápidamente al próximo elemento, no necesariamente mantener una lista completamente ordenada. heapq en Python resuelve este problema mediante un heap binario almacenado en una lista normal.

En esta guía aprenderás a crear colas de prioridad, insertar y retirar elementos, transformar listas existentes, seleccionar valores mínimos o máximos y tratar prioridades repetidas. El tema se relaciona con el módulo collections, la guía de sort y sorted, los métodos de las listas, las funciones en Python y las recomendaciones para mejorar scripts lentos.

Qué es un heap

Un heap es un árbol binario representado dentro de una lista. En el min-heap utilizado por defecto, cada nodo padre tiene un valor menor o igual que el de sus hijos. Esta regla recibe el nombre de invariante del heap. No significa que toda la lista esté ordenada. Solamente garantiza que el elemento mínimo se encuentra en el índice cero.

Para una posición k, los hijos aparecen en 2*k + 1 y 2*k + 2. La representación con listas evita crear objetos de árbol separados y mantiene la estructura compacta. La documentación oficial de heapq explica esta organización, las operaciones disponibles y las APIs de heap mínimo y máximo.

Por qué usar heapq en lugar de ordenar todo

Ordenar n elementos cuesta normalmente O(n log n). Si el programa añade una sola tarea y únicamente necesita conocer el menor elemento, volver a ordenar toda la lista repite trabajo. En un heap, insertar o retirar la raíz cuesta O(log n), mientras que consultar el valor mínimo mediante heap[0] cuesta O(1).

Transformar una lista completa con heapify() es una operación lineal, O(n). Cuando los valores iniciales ya están disponibles, suele ser mejor aplicar heapify una vez que insertar cada elemento individualmente.

Convertir una lista con heapify

import heapq

numeros = [18, 4, 12, 7, 2, 30, 9]
heapq.heapify(numeros)

print(numeros)
print(numeros[0])  # valor mínimo

La lista resultante puede no parecer ordenada, pero conserva la invariante. No conviene comprobar una distribución interna exacta. El contrato importante es que el índice cero contiene el menor elemento y que las operaciones del módulo mantienen la estructura válida.

Insertar y retirar elementos

Utiliza heappush() para insertar y heappop() para retirar el menor valor:

import heapq

cola = []
heapq.heappush(cola, 20)
heapq.heappush(cola, 5)
heapq.heappush(cola, 12)

while cola:
    siguiente = heapq.heappop(cola)
    print(siguiente)

Los números se muestran en orden creciente. Para consultar el menor valor sin retirarlo, lee cola[0]. Llamar a heappop() con una lista vacía produce IndexError, por lo que debes comprobar el estado o controlar la excepción.

Crear una cola de prioridad con tuplas

Una cola real suele almacenar una prioridad junto con la tarea. Python compara las tuplas de izquierda a derecha, así que la prioridad debe ocupar la primera posición:

import heapq

cola = []
heapq.heappush(cola, (3, "generar informe"))
heapq.heappush(cola, (1, "restaurar servicio"))
heapq.heappush(cola, (2, "responder cliente"))

prioridad, tarea = heapq.heappop(cola)
print(prioridad, tarea)

En esta convención, un número menor representa una mayor urgencia. Otra regla también es válida, pero debe estar documentada y mantenerse en todo el proyecto.

Tratar prioridades iguales

Cuando dos prioridades son iguales, la comparación continúa con el segundo elemento. Las cadenas pueden ordenarse, pero los diccionarios, objetos personalizados o valores mezclados pueden generar TypeError. Además, quizá quieras conservar el orden de llegada. Un contador creciente resuelve ambos problemas:

import heapq
from itertools import count

secuencia = count()
cola = []

heapq.heappush(cola, (2, next(secuencia), {"id": 101}))
heapq.heappush(cola, (2, next(secuencia), {"id": 102}))

prioridad, orden, tarea = heapq.heappop(cola)
print(tarea)

El número de secuencia funciona como desempate estable. Como cada valor es único, Python nunca necesita comparar directamente las tareas.

Usar una dataclass para los elementos

Una clase puede expresar mejor el formato. El campo que contiene la tarea debe ignorarse durante la comparación:

from dataclasses import dataclass, field
from typing import Any
import heapq

@dataclass(order=True)
class ElementoPriorizado:
    prioridad: int
    elemento: Any = field(compare=False)

cola = [
    ElementoPriorizado(4, "copia de seguridad"),
    ElementoPriorizado(1, "alerta crítica"),
]
heapq.heapify(cola)
print(heapq.heappop(cola).elemento)

Este patrón mejora la legibilidad cuando las entradas priorizadas pasan por diferentes módulos o capas.

Actualizar o eliminar tareas pendientes

Modificar una posición arbitraria puede romper la invariante. Buscar una tarea también exige recorrer la lista. Una solución habitual mantiene un diccionario que apunta a la entrada actual de cada tarea. Cuando cambia la prioridad, la entrada anterior se marca como eliminada y se añade una nueva. La retirada ignora las entradas antiguas.

import heapq
from itertools import count

ELIMINADA = object()
heap = []
entradas = {}
secuencia = count()

def agregar(tarea, prioridad):
    if tarea in entradas:
        eliminar(tarea)
    entrada = [prioridad, next(secuencia), tarea]
    entradas[tarea] = entrada
    heapq.heappush(heap, entrada)

def eliminar(tarea):
    entrada = entradas.pop(tarea)
    entrada[2] = ELIMINADA

def retirar():
    while heap:
        prioridad, _, tarea = heapq.heappop(heap)
        if tarea is not ELIMINADA:
            del entradas[tarea]
            return tarea, prioridad
    raise KeyError("cola de prioridad vacía")

La eliminación diferida conserva la eficiencia de las operaciones principales y evita reconstruir el heap en cada cambio.

heappushpop y heapreplace

Las funciones combinadas son útiles para heaps de tamaño fijo. heappushpop(heap, item) inserta el valor nuevo y retira el menor en una sola operación optimizada. Devuelve el menor entre el elemento nuevo y la raíz anterior, dejando el mayor en la estructura.

heapreplace(heap, item) retira primero la raíz existente y después inserta el nuevo valor. El tamaño no cambia, pero el heap no puede estar vacío. La diferencia es importante cuando conservas los tres valores más altos observados:

import heapq

mayores = [8, 2, 15]
heapq.heapify(mayores)

for valor in [3, 21, 7]:
    if valor > mayores[0]:
        heapq.heapreplace(mayores, valor)

print(sorted(mayores, reverse=True))

El heap conserva solamente tres valores. Su raíz es el menor candidato actual y puede sustituirse cuando aparece un valor mejor.

Encontrar los valores mayores y menores

nlargest() y nsmallest() devuelven una cantidad limitada de elementos y aceptan una función key:

import heapq

productos = [
    {"nombre": "A", "precio": 80},
    {"nombre": "B", "precio": 25},
    {"nombre": "C", "precio": 110},
    {"nombre": "D", "precio": 45},
]

mas_caros = heapq.nlargest(2, productos, key=lambda p: p["precio"])
mas_barato = heapq.nsmallest(1, productos, key=lambda p: p["precio"])
print(mas_caros)
print(mas_barato)

Estas funciones son adecuadas cuando n es pequeño en comparación con el conjunto. Si necesitas una gran parte de los elementos, sorted() puede ser más eficiente y claro. Para un único resultado, utiliza min() o max().

Heap máximo en Python 3.14

Durante años, la técnica común para simular un max-heap consistía en insertar números negativos. Python 3.14 añadió funciones explícitas: heapify_max(), heappush_max(), heappop_max(), heappushpop_max() y heapreplace_max().

import heapq

valores = [4, 19, 7, 12]
heapq.heapify_max(valores)
print(heapq.heappop_max(valores))  # 19

Los proyectos compatibles con versiones anteriores todavía pueden negar prioridades numéricas, pero la convención debe estar bien documentada. Las APIs con sufijo _max expresan la intención directamente y evitan errores de signo.

Combinar secuencias ordenadas

heapq.merge() combina varias entradas ya ordenadas y devuelve un iterador. A diferencia de concatenar todos los valores y llamar a sorted, puede procesar flujos sin cargar el conjunto completo en memoria:

import heapq

log_a = [1, 4, 8]
log_b = [2, 3, 10]

for valor in heapq.merge(log_a, log_b):
    print(valor)

Es útil para archivos de registro, exportaciones paginadas y fuentes temporales. Todas las entradas deben usar la misma dirección de ordenación.

heapq o queue.PriorityQueue

heapq ofrece funciones que operan sobre una lista. No incluye bloqueos ni espera entre productores y consumidores. Cuando varias threads comparten trabajo, la clase descrita en la documentación de queue.PriorityQueue ofrece sincronización, límite opcional de capacidad y operaciones bloqueantes.

Elige heapq para algoritmos locales, procesos de una sola thread o código que ya controla la sincronización. Elige PriorityQueue para una cola segura entre threads. Las aplicaciones asíncronas también pueden usar asyncio.PriorityQueue.

Ejemplo de planificador de tareas

Un planificador puede ordenar trabajos por el instante de ejecución. La fecha más cercana siempre sale primero:

import heapq
from datetime import datetime, timedelta

agenda = []
ahora = datetime.now()

heapq.heappush(agenda, (hora + timedelta(minutes=10), "enviar informe"))
heapq.heappush(agenda, (hora + timedelta(minutes=2), "actualizar caché"))
heapq.heappush(agenda, (hora + timedelta(minutes=5), "comprobar importaciones"))

momento, tarea = heapq.heappop(agenda)
print(momento, tarea)

Un planificador de producción también necesita persistencia, zonas horarias, reintentos, cancelación y control de workers duplicados. El heap resuelve el orden de las tareas, no todo el ciclo de ejecución.

Errores frecuentes

• Suponer que toda la lista está ordenada porque heap[0] es el menor valor.
• Usar append() en lugar de heappush() después de convertir la lista.
• Eliminar posiciones arbitrarias y romper la invariante.
• Comparar tareas incompatibles cuando existen prioridades iguales.
• Utilizar valores negativos para un max-heap sin documentarlo.
• Aplicar nlargest a casi todos los elementos cuando ordenar sería más simple.
• Compartir la lista entre threads sin protección.

Cómo probar una cola de prioridad

No compruebes una forma interna exacta, porque diferentes listas pueden representar heaps válidos. Prueba el comportamiento visible: el orden de retirada, los empates, las actualizaciones, la eliminación diferida y los errores de cola vacía.

def test_orden_prioridad():
    cola = []
    heapq.heappush(cola, (3, "baja"))
    heapq.heappush(cola, (1, "alta"))
    heapq.heappush(cola, (2, "media"))

    assert heapq.heappop(cola)[1] == "alta"
    assert heapq.heappop(cola)[1] == "media"
    assert heapq.heappop(cola)[1] == "baja"

Buenas prácticas

• Define si un número menor representa una prioridad mayor.
• Usa un contador para conservar el orden de llegada.
• Encapsula el heap en una clase cuando necesites actualizar o cancelar.
• Aplica heapify() a colecciones existentes.
• Prefiere operaciones combinadas en heaps con tamaño limitado.
• Mide antes de sustituir una ordenación sencilla.
• Documenta la versión mínima de Python al usar APIs de max-heap.

Conclusión

heapq en Python permite acceder rápidamente a elementos prioritarios sin ordenar toda la colección después de cada modificación. Con heapify, heappush, heappop, operaciones combinadas, selección de extremos, mezcla de flujos y APIs de heap máximo, el módulo sirve para algoritmos compactos y colas de tareas prácticas.

La clave consiste en preservar la invariante y diseñar correctamente el formato de cada entrada. Para casos simples basta una tupla con prioridad y tarea. Los empates, cambios y cancelaciones requieren un contador, un diccionario auxiliar y eliminación diferida. Con estos patrones, la cola permanece eficiente, predecible y fácil de probar.