python / expert
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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import asyncioasync 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
1
async with asyncio.TaskGroup() as tg:
Erstellt einen Geltungsbereich, der auf den Abschluss aller Aufgaben wartet, bevor er beendet wird.
2
except* ValueError as eg:
Verwendet die neue 'except*'-Syntax, um mehrere Ausnahmen zu behandeln, die in einer ExceptionGroup verpackt sind.