python async library - recommended way to wait for task completion

When using async library, it is always good to use TaskGroup to manage execution of different tasks where we use create_task to create async task. 

import asyncio

import time

async def worker(task_id: int, delay: int):

    print(f"Task {task_id} started (sleeping {delay}s)...")

    await asyncio.sleep(delay)

    print(f"Task {task_id} finished after {delay}s!")

    return f"Result {task_id}"

async def main():

    start_time = time.time()

   

    # TaskGroup manages execution and waits until all tasks finish

    async with asyncio.TaskGroup() as tg:

        t1 = tg.create_task(worker(1, 1))

        t2 = tg.create_task(worker(2, 2))

        t3 = tg.create_task(worker(3, 10))

   

    # All tasks are guaranteed complete after exiting the context block

    print("Results:", t1.result(), t2.result(), t3.result())

    print(f"Total time elapsed: {time.time() - start_time:.2f} seconds")

if __name__ == "__main__":

    asyncio.run(main())

   

Remember don't run async.io.run multiple times.

And to cancel a task where can use task.cancel as shown below :-

import asyncio

import signal

import sys

async def worker():

    try:

        while True:

            print("Worker processing...")

            await asyncio.sleep(1)

    except asyncio.CancelledError:

        print("Worker task received cancellation request.")

async def main():

    # Spawn background task

    task = asyncio.create_task(worker())

   

    # Run for 3 seconds, then cancel task cleanly

    await asyncio.sleep(3)

    task.cancel()

    await task

if __name__ == "__main__":

    try:

        asyncio.run(main())

    except KeyboardInterrupt:

        print("\nProcess interrupted by user (Ctrl+C). Exiting cleanly.")

        sys.exit(0)