Skip to content

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érationFonctionComplexité temporelle
Insérer un élémentheapq.heappush(heap, item)O(log n)
Extraire le plus petitheapq.heappop(heap)O(log n)
Insérer puis extraireheapq.heappushpop(heap, item)O(log n)
Extraire puis insérerheapq.heapreplace(heap, item)O(log n)
Créer un tas depuis une listeheapq.heapify(list)O(n)
N plus petits élémentsheapq.nsmallest(n, iterable)O(n log k)
N plus grands élémentsheapq.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.8

Quand utiliser nlargest/nsmallest vs sorted()

ScénarioMeilleure approchePourquoi
N est 1min() / max()O(n), le plus simple
N est petit par rapport au totalnlargest / nsmallestO(n log k), moins de mémoire
N est proche de la longueur totalesorted(iterable)[:n]O(n log n), mais rapide en pratique
Tous les éléments triés nécessairessorted()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 1

File 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 logs

Le 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.

📚