Python heapq : Files de priorité et opérations de tas simplifiées
Updated on
Vous devez traiter des éléments par priorité -- des tâches dans un planificateur de travaux, des événements dans une simulation ou le chemin le plus court dans un graphe. Une liste triée semble le choix évident, mais l'insertion dans une liste triée est O(n), ce qui devient un goulot d'étranglement lorsque vous traitez des millions d'éléments. Retrier toute la liste à chaque ajout d'élément est encore pire à O(n log n).
Le coût s'accumule rapidement. Un planificateur de travaux qui retrie sa file à chaque insertion gaspille la majeure partie de son temps en tâches administratives. Un algorithme de recherche de chemin qui scanne une liste complète pour trouver la distance minimale à chaque étape transforme une opération linéaire en une opération quadratique.
Le module heapq de Python fournit une file de priorité basée sur un tas offrant une insertion O(log n) et une extraction O(log n) du plus petit élément. Il inclut également des fonctions optimisées pour trouver les N plus grands ou plus petits éléments de n'importe quel itérable sans trier toute la collection. Ce guide couvre chaque fonction de heapq avec des exemples pratiques.
Qu'est-ce qu'un Tas ?
Un tas est un arbre binaire où chaque nœud parent est inférieur ou égal à ses enfants (tas minimum). Le heapq de Python implémente un tas minimum en utilisant une liste régulière, où le plus petit élément est toujours à l'index 0.
Les opérations clés et leurs complexités temporelles :
| Opération | Fonction | Complexité temporelle |
|---|---|---|
| Insérer un élément | heapq.heappush(heap, item) | O(log n) |
| Extraire le plus petit | heapq.heappop(heap) | O(log n) |
| Insérer puis extraire | heapq.heappushpop(heap, item) | O(log n) |
| Extraire puis insérer | heapq.heapreplace(heap, item) | O(log n) |
| Créer un tas depuis une liste | heapq.heapify(list) | O(n) |
| N plus petits éléments | heapq.nsmallest(n, iterable) | O(n log k) |
| N plus grands éléments | heapq.nlargest(n, iterable) | O(n log k) |
Opérations de base du Tas
Créer et utiliser un Tas
import heapq
# Start with an empty heap (just a regular list)
heap = []
# Push items
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 2)
print(heap) # [1, 2, 3, 5] (heap order, not sorted)
print(heap[0]) # 1 (smallest element, always at index 0)
# Pop smallest item
smallest = heapq.heappop(heap)
print(smallest) # 1
print(heap) # [2, 5, 3]Convertir une liste en Tas
import heapq
data = [5, 3, 8, 1, 2, 7, 4, 6]
heapq.heapify(data) # In-place, O(n)
print(data) # [1, 2, 4, 3, 5, 7, 8, 6] (heap order)
# Pop elements in sorted order
sorted_data = []
while data:
sorted_data.append(heapq.heappop(data))
print(sorted_data) # [1, 2, 3, 4, 5, 6, 7, 8]heappushpop et heapreplace
Ces opérations combinées sont plus efficaces que des appels push/pop séparés :
import heapq
heap = [1, 3, 5, 7]
heapq.heapify(heap)
# heappushpop: push first, then pop smallest
result = heapq.heappushpop(heap, 2)
print(result) # 1 (pushed 2, popped 1)
print(heap) # [2, 3, 5, 7]
# heapreplace: pop smallest first, then push
result = heapq.heapreplace(heap, 10)
print(result) # 2 (popped 2, pushed 10)
print(heap) # [3, 7, 5, 10]nlargest et nsmallest
Ces fonctions trouvent les N plus grands ou plus petits éléments sans trier toute la collection.
import heapq
scores = [85, 92, 78, 95, 88, 72, 91, 83, 97, 69]
# Top 3 scores
top_3 = heapq.nlargest(3, scores)
print(top_3) # [97, 95, 92]
# Bottom 3 scores
bottom_3 = heapq.nsmallest(3, scores)
print(bottom_3) # [69, 72, 78]Avec une fonction clé
Les deux acceptent un paramètre key pour une comparaison personnalisée :
import heapq
students = [
{'name': 'Alice', 'gpa': 3.9},
{'name': 'Bob', 'gpa': 3.5},
{'name': 'Charlie', 'gpa': 3.8},
{'name': 'Diana', 'gpa': 3.95},
{'name': 'Eve', 'gpa': 3.2},
]
top_students = heapq.nlargest(3, students, key=lambda s: s['gpa'])
for s in top_students:
print(f"{s['name']}: {s['gpa']}")
# Diana: 3.95
# Alice: 3.9
# Charlie: 3.8Quand utiliser nlargest/nsmallest vs sorted()
| Scénario | Meilleure approche | Pourquoi |
|---|---|---|
| N est 1 | min() / max() | O(n), le plus simple |
| N est petit par rapport au total | nlargest / nsmallest | O(n log k), moins de mémoire |
| N est proche de la longueur totale | sorted(iterable)[:n] | O(n log n), mais rapide en pratique |
| Tous les éléments triés nécessaires | sorted() | Le tri complet est approprié |
Implémenter un Tas Maximum
Le heapq de Python ne fournit qu'un tas minimum. Pour créer un tas maximum, inversez les valeurs :
import heapq
# Max-heap using negation
max_heap = []
for val in [5, 1, 3, 8, 2]:
heapq.heappush(max_heap, -val)
# Pop largest items
while max_heap:
print(-heapq.heappop(max_heap), end=' ')
# 8 5 3 2 1File de priorité avec des objets personnalisés
Pour des objets qui ne sont pas directement comparables, utilisez des tuples (priority, counter, item) :
import heapq
from dataclasses import dataclass
@dataclass
class Task:
name: str
description: str
# Use (priority, tie_breaker, task) tuples
task_queue = []
counter = 0
def add_task(priority, task):
global counter
heapq.heappush(task_queue, (priority, counter, task))
counter += 1
def get_next_task():
if task_queue:
priority, _, task = heapq.heappop(task_queue)
return task
return None
add_task(3, Task("Low", "Clean up logs"))
add_task(1, Task("Critical", "Fix production bug"))
add_task(2, Task("Medium", "Update docs"))
add_task(1, Task("Critical", "Patch security hole"))
while task_queue:
task = get_next_task()
print(f"{task.name}: {task.description}")
# Critical: Fix production bug
# Critical: Patch security hole
# Medium: Update docs
# Low: Clean up logsLe compteur sert de départage pour que les éléments de même priorité soient retournés dans l'ordre d'insertion (FIFO).
Cas d'utilisation pratiques
Fusionner des itérables triés
import heapq
list1 = [1, 4, 7, 10]
list2 = [2, 5, 8, 11]
list3 = [3, 6, 9, 12]
merged = list(heapq.merge(list1, list2, list3))
print(merged) # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]heapq.merge() est économe en mémoire -- il ne charge pas toutes les listes en mémoire simultanément.
Médiane glissante
import heapq
class RunningMedian:
"""Track the median of a stream of numbers."""
def __init__(self):
self.low = [] # max-heap (negated values)
self.high = [] # min-heap
def add(self, num):
heapq.heappush(self.low, -num)
heapq.heappush(self.high, -heapq.heappop(self.low))
if len(self.high) > len(self.low):
heapq.heappush(self.low, -heapq.heappop(self.high))
def median(self):
if len(self.low) > len(self.high):
return -self.low[0]
return (-self.low[0] + self.high[0]) / 2
rm = RunningMedian()
for num in [2, 1, 5, 7, 2, 0, 5]:
rm.add(num)
print(f"Added {num}, median = {rm.median()}")Plus court chemin de Dijkstra
import heapq
def dijkstra(graph, start):
"""Find shortest paths from start to all nodes."""
distances = {node: float('inf') for node in graph}
distances[start] = 0
pq = [(0, start)]
while pq:
dist, node = heapq.heappop(pq)
if dist > distances[node]:
continue
for neighbor, weight in graph[node]:
new_dist = dist + weight
if new_dist < distances[neighbor]:
distances[neighbor] = new_dist
heapq.heappush(pq, (new_dist, neighbor))
return distances
graph = {
'A': [('B', 1), ('C', 4)],
'B': [('C', 2), ('D', 5)],
'C': [('D', 1)],
'D': [],
}
print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}Analyser les performances du Tas dans Jupyter
Lors de la comparaison d'algorithmes basés sur les tas avec des alternatives ou du profilage des performances des files de priorité, l'expérimentation interactive accélère le processus. RunCell (opens in a new tab) est un agent IA pour Jupyter qui peut vous aider à benchmarker les opérations de tas, visualiser les caractéristiques de performance et optimiser vos choix de structures de données de manière interactive.
FAQ
Qu'est-ce que heapq en Python ?
heapq est un module de la bibliothèque standard qui implémente un tas minimum en utilisant une liste Python régulière. Il fournit des fonctions pour insérer et extraire des éléments en temps O(log n) tout en maintenant la propriété du tas (le plus petit élément toujours à l'index 0). C'est le moyen standard d'implémenter des files de priorité en Python.
Comment créer un tas maximum en Python ?
Le heapq de Python ne supporte que les tas minimum. Pour simuler un tas maximum, inversez les valeurs lors de l'insertion et inversez à nouveau lors de l'extraction : heapq.heappush(heap, -value) pour insérer, -heapq.heappop(heap) pour obtenir le plus grand. Pour des objets complexes, inversez la priorité dans un tuple : (-priority, counter, item).
Quand dois-je utiliser heapq vs sorted() ?
Utilisez heapq.nlargest(n, data) ou heapq.nsmallest(n, data) quand vous n'avez besoin que des top N éléments et que N est petit par rapport à la taille totale des données. Utilisez sorted() quand vous avez besoin de tous les éléments ordonnés ou quand N est proche de la taille totale. Utilisez min()/max() quand N est 1.
heapq est-il thread-safe ?
Non. Les appels individuels de heappush et heappop ne sont pas atomiques. Si plusieurs threads accèdent au même tas, vous avez besoin d'une synchronisation externe. Pour des files de priorité thread-safe, utilisez queue.PriorityQueue, qui encapsule heapq avec un verrouillage approprié.
Quelle est la complexité temporelle des opérations de heapq ?
heappush et heappop sont tous deux O(log n). heapify convertit une liste en tas en O(n). nlargest(k, data) et nsmallest(k, data) s'exécutent en O(n log k) où n est la taille des données et k le nombre d'éléments demandés.
Conclusion
Le module heapq de Python est l'outil standard pour les files de priorité et la sélection top-N. Son approche basée sur les tas offre des opérations push et pop en O(log n), ce qui le rend dramatiquement plus rapide que le maintien d'une liste triée. Utilisez heappush/heappop pour les modèles de files de priorité, nlargest/nsmallest pour la sélection top-N, merge pour combiner des séquences triées et heapify quand vous avez une liste existante à convertir.