How do you cancel or add a timeout to an asyncio task safely?

Timeouts with asyncio.wait_for

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
    except TimeoutError:
        print("timed out after 2 seconds")

asyncio.run(main())

wait_for runs the coroutine as a task and races it against the timeout; if the timeout elapses first, it cancels the task internally and raises TimeoutError (in Python 3.11+; asyncio.TimeoutError, an alias of the same, in earlier versions) to the caller.

Python 3.11+'s asyncio.timeout(): a context-manager alternative

async def main():
    try:
        async with asyncio.timeout(2):
            await slow_operation()
    except TimeoutError:
        print("timed out")

asyncio.timeout() applies a deadline to an entire async with block (potentially covering multiple awaits), which is more flexible than wait_for when you want one timeout to cover several sequential operations rather than wrapping each individually.

Explicit cancellation with task.cancel()

async def worker():
    try:
        await asyncio.sleep(100)
    except asyncio.CancelledError:
        print("cleaning up before exiting")
        raise             # IMPORTANT: re-raise so the cancellation actually completes

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(1)
    task.cancel()          # requests cancellation
    await task              # propagates CancelledError here, after cleanup runs

task.cancel() doesn't stop the task immediately — it arranges for CancelledError to be raised inside the task at its next await point. The task can catch that to run cleanup (closing a connection, releasing a lock), but must re-raise it afterward; swallowing CancelledError silently prevents the cancellation from actually taking effect and can leave .cancel() callers waiting forever.

Why you must never blanket-catch CancelledError

async def bad_worker():
    try:
        await asyncio.sleep(100)
    except Exception:        # BUG: in Python 3.8+, CancelledError is a
        pass                  # BaseException subclass, NOT caught here --
                                # but a bare `except:` WOULD catch it and break cancellation

Since Python 3.8, asyncio.CancelledError inherits from BaseException (not Exception), specifically so that ordinary except Exception: blocks don't accidentally swallow it — but a bare except: still would, so it's a rule worth stating explicitly: always let cancellation propagate unless you're deliberately intercepting it to clean up, and always re-raise afterward.

Interview-ready summary: Use asyncio.wait_for/asyncio.timeout() for deadline-based cancellation, and task.cancel() for explicit cancellation — both work by raising CancelledError inside the task at its next await. Treat CancelledError as something to clean up around, never to swallow, since eating it silently breaks the cancellation contract for whoever is waiting on the task.