Dans asyncio, il est possible d'inspecter les tâches en cours d'exécution au sein de la boucle d'événements. Cela se fait en obtanant des objets asyncio.Task pour la tâche actuelle et toutes les tâches actives.

Accès à la tâche en cours d'exécution

La fonction asyncio.current_task() permet de récupérer la tâche qui s'exécute actuellement. Elle retourne un objet Task représentant cette tâche.

# Obtenir la tâche courante
active_task = asyncio.current_task()

Cet objet peut correspondre à la coroutine principale passée à asyncio.run() ou à une tâche créée via asyncio.create_task(). Si une coroutine s'exécute indépendamment, asyncio.current_task() retourne la tâche parente, pas la coroutine elle-même. Cette méthode est utile pour accéder aux métadonnées de la tâche, comme son nom pour le journalisation.

Voici un exemple où la coroutine principale récupère sa propre tâche et affiche ses détails, illustrant que toute coroutine est encapsulée dans une tâche dans asyncio.

import asyncio

async def entry_point():
    print('La coroutine principale a démarré')
    current = asyncio.current_task()
    print(current)

asyncio.run(entry_point())

L'exécution produit la sortie suivante, montrant que la tâche par défaut nommée "Task-1" exécute la coroutine entry_point().

La coroutine principale a démarré
<task ...="" at="" coro="<entry_point()" name="Task-1" pending="" running=""> cb=[_run_until_complete_cb() at ...]>
</task>

Récupération de toutes les tâches planifiées

Pour inspecter l'état global du programme asyncio, on peut obtenir l'ensemble de toutes les tâches planifiées ou en cours d'exécution avec asyncio.all_tasks().

# Obtenir toutes les tâches
all_tasks_set = asyncio.all_tasks()

Cette fonction retourne un ensemble contanant toutes les tâchees qui n'ont pas encore terminé, y compris la tâche courante. Un exemple pratique consiste à créer plusieurs tâches, puis à lister leurs propriétés.

import asyncio

async def worker(id_number):
    print(f'Le travailleur {id_number} est actif')
    await asyncio.sleep(1)

async def main():
    print('Début du programme principal')
    tasks_created = [asyncio.create_task(worker(i)) for i in range(10)]
    await asyncio.sleep(0.1)  # Permettre aux tâches de commencer
    all_tasks = asyncio.all_tasks()
    for t in all_tasks:
        print(f'> Nom: {t.get_name()}, Coroutine: {t.get_coro()}')
    for t in tasks_created:
        await t

asyncio.run(main())

Cet exemple crée dix tâches qui exécutent la coroutine worker(), puis liste toutes les tâches, incluant la tâche principale. La sortie montre les noms et les coroutines associées, démontrant que asyncio.all_tasks() fournit une vue complète des tâches actives.