Pattern 10: Cancellation & Cleanup

The Problem: Resource Leaks

Scenario: A long-running task opens a connection, starts streaming data, or acquires a lock.

User cancels: Ctrl+C or timeout. The task is killed but cleanup code never runs.

Result: Leaked connections, dangling locks, incomplete writes.

Solution: Use `try/finally`, `async with`, or `TaskGroup` to ensure cleanup happens.

Cancellation Flow

sequenceDiagram participant User participant Task as Task participant Resource as Resource User->>Task: Start task Task->>Resource: Acquire (lock, connection, file) Resource-->>Task: OK Note over Task: Doing work... User->>Task: Cancel (Ctrl+C, timeout) Task->>Task: CancelledError raised Task->>Resource: finally block runs Task->>Resource: Release resource Resource-->>Task: OK Task-->>User: Cleanup complete! Note over Resource: ✅ No leaks

Key rule: Every `acquire` must have a `release` in a `finally` block or async context manager.

Three Patterns for Cleanup

1. try/finally (Manual)

async def task_with_manual_cleanup():
    resource = None
    try:
        resource = await acquire_lock()
        # Do work
        await asyncio.sleep(10)
    finally:
        if resource:
            await release_lock(resource)
            print("Cleanup done!")

async def main():
    try:
        await task_with_manual_cleanup()
    except asyncio.CancelledError:
        print("Task was cancelled, cleanup ran")

2. async with (Automatic)

async def task_with_context_manager():
    async with acquire_lock() as lock:
        # Lock is acquired here
        await asyncio.sleep(10)
        # Lock is released automatically in __aexit__
        # Even if task is cancelled!

async def main():
    try:
        await task_with_context_manager()
    except asyncio.CancelledError:
        print("Lock released automatically!")

3. TaskGroup (3.11+, Recommended)

async def worker(id):
    async with acquire_connection() as conn:
        # Connection acquired
        await asyncio.sleep(10)
        # Connection released on cancel

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            for i in range(5):
                tg.create_task(worker(i))
    except asyncio.CancelledError:
        # All tasks cancelled
        # All cleanup code (finally blocks) runs
        print("All tasks cleaned up!")

asyncio.run(main())

Real-World: Graceful Shutdown

import signal

async def watch_stream():
    async with get_connection() as conn:
        async for message in conn.stream():
            print(f"Message: {message}")

async def main():
    task = asyncio.create_task(watch_stream())

    # Catch Ctrl+C
    loop = asyncio.get_event_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(
            sig,
            lambda t=task: t.cancel()
        )

    try:
        await task
    except asyncio.CancelledError:
        print("Gracefully cancelled")

asyncio.run(main())

FastAPI: Request Cancellation

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.get("/long-task")
async def long_task(background_tasks: BackgroundTasks):
    async def background_work():
        try:
            async with get_db_connection() as conn:
                for i in range(100):
                    await conn.query(...)
                    await asyncio.sleep(1)
        except asyncio.CancelledError:
            # User cancelled request → cleanup runs
            print("Long task cancelled, connection closed")
            raise

    background_tasks.add_task(background_work)
    return {"status": "started"}