`Structured Concurrency cu asyncio.TaskGroup în Python 3.11+` în Python

Programare asincronă Python 3.11+

Ghid complet pentru Structured Concurrency în Python 3.11+: lansarea și sincronizarea sigură a sarcinilor asincrone cu asyncio.TaskGroup și eliminarea sarcinilor orfane.

Sintaxă

async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(descarca_a())
    task2 = tg.create_task(descarca_b())

Înainte de Python 3.11, executarea mai multor corutine concurente se baza pe asyncio.gather(). Deși popular, gather() prezenta o problemă majoră de fiabilitate: dacă una dintre sarcini eșua cu o excepție, celelalte sarcini lansate continuau să ruleze în fundal ca sarcini orfane (leaked tasks), consumând resurse de rețea și memorie fără a mai fi supravegheate de nimeni.

Introdus în Python 3.11 pe baza conceptului de Structured Concurrency, managerul de context asyncio.TaskGroup garantează că toate sarcinile lansate în interiorul său sunt complet finalizate sau anulate curat înainte de ieșirea din bloc.

Execuție concurentă sigură cu asyncio.TaskGroup

import asyncio

async def preia_metadate(articol_id: int):
    await asyncio.sleep(0.5)
    return {"id": articol_id, "titlu": f"Articol {articol_id}"}

async def preia_comentarii(articol_id: int):
    await asyncio.sleep(0.8)
    return [f"Comentariu 1 pentru {articol_id}", f"Comentariu 2"]

async def main():
    async with asyncio.TaskGroup() as tg:
        # Lansăm două sarcini paralele în interiorul grupului:
        task_meta = tg.create_task(preia_metadate(101))
        task_comm = tg.create_task(preia_comentarii(101))
        
    # În acest punct, ambele task-uri sunt GARANTAT finalizate!
    print("Metadate:", task_meta.result())
    print("Comentarii:", task_comm.result())

asyncio.run(main())

Ce se întâmplă când o sarcină aruncă o eroare?

Dacă orice sarcină lansată într-un TaskGroup eșuează:

Grupul anulează automat în cascadă toate celelalte sarcini active care încă rulează (prin transmiterea unei cereri de cancel()).
Așteaptă ca toate sarcinile anulate să se oprească curat.
Împachetează toate excepțiile apărute într-un obiect de tip ExceptionGroup, permițând tratarea selectivă a fiecărei erori.

Tratarea erorilor din TaskGroup cu sintaxa except*

async def sarcina_cu_probleme():
    await asyncio.sleep(0.1)
    raise ValueError("Date corupte!")

async def sarcina_lenta():
    await asyncio.sleep(10.0)

async def test_erori():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(sarcina_cu_probleme())
            tg.create_task(sarcina_lenta()) # Va fi anulată instantaneu când prima dă eroare!
    except* ValueError as eg:
        print("Eroare de valoare prinsă din grup:", eg.exceptions)

asyncio.run(test_erori())