Python asyncio Recipes/Cancel cleanly on shutdown

Maintain and cancel outstanding tasks at shutdown time.

Section: Retries and Resilience

Cancel cleanly on shutdown

python
python
import asyncio

async def worker():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("worker shutting down")
        raise

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.2)
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)

asyncio.run(main())
Explanation

Use this pattern to ensure graceful shutdown in services and daemons.

Learn the surrounding workflow

Compare similar commands or jump into common fixes when this command is part of a bigger troubleshooting path.

Related commands

Same sheet · prioritizing Retries and Resilience
Retry an async operation
Retry with simple exponential backoff.
OpenIn sheetpythonsame section
Bound a worker pool with Semaphore
Throttle async work to a safe level.
Inspect running tasks
List currently active tasks in the running loop.
Process input in batches
Break large workloads into fixed-size async chunks.
Enable asyncio debug mode
Turn on extra diagnostics while developing.