Project: Async Task Scheduler

Python 3.14 · ✓ verified by execution on 2026-07-31

You’ve mastered coroutines, gathering tasks, type hinting, and custom exceptions. Now it’s time to build a robust Async Task Scheduler.

In production systems, you often have a pool of diverse tasks to run concurrently (e.g., fetching from multiple APIs). Some might fail instantly. Some might hang forever.

A resilient scheduler must:

Run everything concurrently.
Enforce strict timeouts on every task.
Catch and catalog all exceptions without crashing the main event loop.

Combining wait_for and gather

asyncio.wait_for(coro, timeout) is brilliant for wrapping a single coroutine. asyncio.gather(*coros) is brilliant for running multiple coroutines at once.

When you combine them, you wrap the wait_for calls inside the gather list.

import asyncio

async def fast():
    return "Done!"

async def slow():
    await asyncio.sleep(10)
    return "Done!"

async def failing():
    raise ValueError("DB Crash")

async def main():
    # Wrap each task in a timeout!
    tasks = [
        asyncio.wait_for(fast(), timeout=1.0),
        asyncio.wait_for(slow(), timeout=1.0),
        asyncio.wait_for(failing(), timeout=1.0)
    ]
    
    # We MUST use return_exceptions=True, otherwise the ValueError 
    # or TimeoutError will instantly blow up the gather call.
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    for r in results:
        print(repr(r))

asyncio.run(main())

'Done!'
TimeoutError()
ValueError('DB Crash')

As you can see, results now cleanly contains a mix of success strings and Exception objects! This is exactly what a production scheduler needs to generate a post-mortem report.

Check yourself

What happens if a coroutine inside `asyncio.gather` fails with an Exception, and `return_exceptions=False` (the default)?

Reveal answer

The exception is raised immediately, but the other concurrent tasks continue running in the background. — If return_exceptions=False, the first exception is raised to the caller, but the other tasks in the gather list are NOT automatically cancelled. If you want them cancelled, you must handle the cancellation manually or use TaskGroups (in Python 3.11+).

How does `asyncio.wait_for` handle coroutine cleanup when a timeout occurs?

Reveal answer

It raises an `asyncio.CancelledError` inside the coroutine, allowing it to execute `finally` blocks. — wait_for triggers a graceful cancellation by raising CancelledError inside the running coroutine. This allows your async code to clean up sockets or database connections in a finally block.

Why is it important to capture exceptions instead of letting them bubble up in a concurrent scheduler?

Reveal answer

So that one failing task doesn't crash the entire batch of tasks. — In a scheduler, tasks are often independent. If one task throws an exception and it bubbles up, it crashes the entire gather call, potentially cancelling or abandoning all other healthy tasks in the batch.

Challenges

Challenge 1 +100 XP

Build an Async Task Scheduler! Create a class `Scheduler`. It should have `tasks = []` and `results = {}`. Write `add_task(self, name, coro)` to append to `tasks`. Write `async def run_all(self, timeout)` which runs all tasks concurrently using `asyncio.gather(..., return_exceptions=True)`. Each task should be wrapped in `asyncio.wait_for(coro, timeout=timeout)`. If it times out, store a `TimeoutError` in `results`. If it fails, store the `Exception` in `results`. Otherwise, store the return value.

Test 1 — expects "True\nTrue\n"

Show solution (0 XP)

import asyncio
class Scheduler:
    def __init__(self):
        self.tasks = []
        self.results = {}
    def add_task(self, name, coro):
        self.tasks.append((name, coro))
    async def _run_wrapped(self, name, coro, timeout):
        try:
            res = await asyncio.wait_for(coro, timeout=timeout)
            self.results[name] = res
        except asyncio.TimeoutError as e:
            self.results[name] = e
        except Exception as e:
            self.results[name] = e
    async def run_all(self, timeout):
        coros = [self._run_wrapped(n, c, timeout) for n, c in self.tasks]
        await asyncio.gather(*coros, return_exceptions=True)
        return self.results

async def quick(): return 42
async def slow(): await asyncio.sleep(2); return 100
async def main():
    s = Scheduler()
    s.add_task('q', quick())
    s.add_task('s', slow())
    res = await s.run_all(0.5)
    print(res['q'] == 42)
    print(isinstance(res['s'], asyncio.TimeoutError))

asyncio.run(main())

Challenge 2 +100 XP

Create a custom exception `TaskFailedError`. Write a coroutine `failing_task` that raises it. Use `asyncio.run()` to execute it and catch the exception, printing 'Caught it!'

Test 1 — expects "Caught it!\n"

Show solution (0 XP)

import asyncio
class TaskFailedError(Exception): pass
async def failing_task():
    raise TaskFailedError('failed')
async def main():
    try:
        await failing_task()
    except TaskFailedError:
        print('Caught it!')
asyncio.run(main())