Verwaltete Nebenläufigkeit mit TaskGroups

Snippet

Verwaltete Nebenläufigkeit mit TaskGroups

Eingeführt in Python 3.11, bieten TaskGroups eine moderne API für nebenläufige Aufgaben. Wenn eine Aufgabe fehlschlägt, werden die anderen automatisch abgebrochen, was ein 'Alles-oder-Nichts'-Ausführungsmuster mit sauberer Ausnahmebehandlung über ExceptionGroups gewährleistet.

snippet.py
python
import asyncio
 
async def fail_task():
    await asyncio.sleep(0.1)
    raise ValueError("Atomic failure")
 
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(asyncio.sleep(1))
            tg.create_task(fail_task())
    except* ValueError as eg:
        for e in eg.exceptions:
            print(f"Caught: {e}")

Erklärung

async with asyncio.TaskGroup() as tg:

Erstellt einen Geltungsbereich, der auf den Abschluss aller Aufgaben wartet, bevor er beendet wird.

except* ValueError as eg:

Verwendet die neue 'except*'-Syntax, um mehrere Ausnahmen zu behandeln, die in einer ExceptionGroup verpackt sind.

Vorheriges Snippet Nächstes Snippet

Verwandt

Aus deiner Bibliothek