Przejdź do treści głównej
🔒 Tryb podglądu. Pierwszych piętnaście lekcji Foundations jest darmowych; ta jest Pro. Rozpocznij 7-dniowy trial, aby odblokować edytor, podpowiedzi AI i resztę kursu. Wymagana karta, anulujesz w dowolnym momencie w Dashboard.Rozpocznij 7-dniowy trial →
← KursySenior Deep-DivesModule 8 · Async correctness patternsasyncio.TaskGroup — structured concurrency (3.11+) replaces gatherpredict132 / 161
+150 XP
Zadanie
📝 **Zadanie:** Przewiduj dokładny wynik 6-liniowy. W grupie zadaniowej pojawiają się trzy zadania: ok-szybko (10 ms, bez awarii), boom (20 ms, podnosi ValueError), ok-slow (50 ms, bez awarii). Grupa zadań przechwytuje wyjątek boomu i drukuje podsumowanie. Prześledź, które linie zadań faktycznie się uruchamiają, a które są anulowane w trakcie snu. 📋 Zaimplementuj powyższą funkcję. Testy uruchamiają się automatycznie. 💡 **Wskazówka:** Jeśli utkniesz, przeczytaj ponownie teorię.
Przewiduj wynik

Przeczytaj kod uważnie

import asyncio


events: list[str] = []


async def task(name: str, work: float, fail: bool = False) -> None:
    events.append(f"{name} start")
    await asyncio.sleep(work)
    if fail:
        events.append(f"{name} raising")
        raise ValueError(f"{name} failed")
    events.append(f"{name} done")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task("ok-fast", 0.01))
            tg.create_task(task("boom", 0.02, fail=True))
            tg.create_task(task("ok-slow", 0.05))
    except* ValueError as eg:
        events.append(f"caught {len(eg.exceptions)} exception(s)")

    for e in events:
        print(e)


asyncio.run(main())

Co wypisze program? Napisz tutaj:

💬 Dyskusja

Zadaj pierwsze pytanie lub podziel się wskazówką.
Zaloguj się aby dołączyć do dyskusji. Czytanie jest darmowe.
Ładowanie dyskusji…