asyncio: Timeouts, Task Cancellation and Graceful Shutdown

import asyncio

async def slow_operation():
    await asyncio.sleep(10)

async def main():
    try:
        async with asyncio.timeout(3.0):   # 3 seconds max
            await slow_operation()
    except TimeoutError:
        print("Timeout exceeded")

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
    except TimeoutError:
        print("Timeout")

async def worker():
    try:
        while True:
            await do_work()
    except asyncio.CancelledError:
        await cleanup()    # release resources
        raise              # IMPORTANT: always re-raise CancelledError

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(5)
    task.cancel()          # send CancelledError into task
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled cleanly")

import asyncio, signal

async def main():
    loop = asyncio.get_running_loop()
    stop = asyncio.Event()

    def handle_signal():
        print("Got SIGINT, shutting down...")
        stop.set()

    loop.add_signal_handler(signal.SIGINT, handle_signal)
    loop.add_signal_handler(signal.SIGTERM, handle_signal)

    # Main work
    tasks = [asyncio.create_task(worker(i)) for i in range(5)]

    await stop.wait()       # wait for signal

    # Cancel all tasks
    for task in tasks:
        task.cancel()

    # Wait for completion with cleanup
    await asyncio.gather(*tasks, return_exceptions=True)
    print("Shut down cleanly")

asyncio.run(main())