Async API Aggregator with Python

Fetch data from 4 APIs concurrently using Python asyncio and httpx.

Introduction

30 Second Summary

When your backend needs data from multiple APIs, fetching them one at a time means your users wait for the sum of all response times. It's like ordering four meals at a restaurant where the kitchen refuses to start the next dish until the previous one is plated.

In this project, you will build a Python script that fetches data from 4 free public APIs concurrently using asyncio. You'll measure the speed difference between sequential and concurrent approaches, then use Python 3.14's newest debugging tools to visualize your async task tree.

What You'll Build

An async API aggregator that pulls weather, country info, random user data, and cryptocurrency prices from 4 live APIs into a single combined report.

By the end of this project, you'll have:

  • A working async aggregator that fetches data from 4 live APIs concurrently and proves a 4-5x speedup over sequential fetching.
  • Hands-on experience with asyncio.gather for simple fan-out concurrency and asyncio.TaskGroup for structured concurrency with automatic error cancellation.
  • Practice handling timeouts, ExceptionGroups, and Python 3.14's call graph introspection for debugging async code.
  • Secret Mission: Add retry logic with exponential backoff and a circuit breaker pattern for handling unreliable APIs.

Are there any prerequisites?

You should be comfortable with basic Python (functions, classes, and context managers). Familiarity with HTTP APIs and JSON responses is helpful but not strictly required.

Set Up Your Project

Before you can fetch data from multiple APIs concurrently, you need a clean Python environment with the right library installed.

In this step, you will create a project folder, set up a virtual environment, install the httpx HTTP library, and scaffold a starter script that confirms everything works.

In this step, get ready to:
  • Create a project folder and virtual environment.
  • Install the httpx library.
  • Create a starter aggregator.py file.
Create your project folder and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This prevents version conflicts between different projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • Open the integrated terminal by pressing Ctrl+` (the backtick key, found above Tab).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir async-aggregator
cd async-aggregator
  • Create a virtual environment by running:
python -m venv venv
  • Confirm the virtual environment was created by running:
ls

You should see a venv folder in the output.

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, so different projects can use different library versions without conflicts.

  • Activate the virtual environment so that any packages you install go into this isolated space.

macOS

  • Run this activation command:
source venv/bin/activate

Windows

  • Run this activation command:
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

  • On macOS, make sure you typed source before the path. Without it, the shell tries to execute the file directly.
  • On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned first, then try the activation command again.
  • Make sure you are inside the async-aggregator folder where you created the venv.

Help me activate my Python virtual environment.

Install httpx

The httpx library provides both synchronous and asynchronous HTTP clients with nearly identical APIs. This makes it perfect for comparing sync vs async approaches side by side.

  • Install httpx by running:
pip install httpx==0.28.1
  • Verify the installation by running:
python -c "import httpx; print(httpx.__version__)"

You should see 0.28.1 printed in the terminal.

Seeing an error during install?

  • Make sure (venv) is showing in your terminal prompt. If it is not, re-run the activation command from the previous substep.
  • If you see a permissions error, confirm you are using the virtual environment's pip and not the system one. The (venv) prefix in your prompt confirms this.

Help me install httpx in my virtual environment.

Create the aggregator.py file

Now you will create the script file that will hold all your aggregator code. You will start with a minimal skeleton that imports the libraries and defines an empty entry point.

  • In VS Code, click File in the top menu bar.
  • Click New File.
  • Save the file immediately with Cmd+S (macOS) or Ctrl+S (Windows).
  • Navigate to your async-aggregator folder on your Desktop, name the file aggregator.py, and click Save.
  • Paste the following starter code into aggregator.py:
import asyncio
import httpx


# Placeholder entry point for our aggregator
async def main():
    pass


# Run the async main function
if __name__ == "__main__":
    asyncio.run(main())

What does this code do?

  • import asyncio brings in Python's built-in library for writing concurrent code with async and await.
  • import httpx loads the HTTP client library you just installed.
  • async def main() defines a coroutine that will become the entry point for all your async logic.
  • asyncio.run(main()) starts the event loop and runs your main() coroutine. This is the standard way to launch async code in Python.
  • Save aggregator.py with Cmd+S (macOS) or Ctrl+S (Windows).
  • Run the script to confirm there are no errors:
python aggregator.py

The command should complete silently with no output and no errors. This confirms your imports are working and the async entry point runs successfully.

Seeing a ModuleNotFoundError?

  • If you see ModuleNotFoundError: No module named 'httpx', your virtual environment is not active. Re-run the activation command, then try python aggregator.py again.
  • Make sure you saved the file inside the async-aggregator folder. Run ls to confirm aggregator.py is listed.

Help me fix a ModuleNotFoundError when running aggregator.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved aggregator.py and that python aggregator.py runs without errors.

ⓧ I'd like to double check the full code

Here is the complete aggregator.py file so far:

import asyncio
import httpx


async def main():
    pass


if __name__ == "__main__":
    asyncio.run(main())

Your environment is ready and your starter script runs cleanly. Next up, you will write four synchronous fetch functions to establish a performance baseline before converting them to async.

Build the Sequential Baseline

Your project skeleton is ready with httpx installed and a placeholder aggregator.py file waiting for code.

Before building anything async, you need a baseline to compare against. In this step, you will fetch data from 4 free public APIs one at a time, measure how long the whole process takes, and format the results into a combined report. This gives you a concrete number to beat when you switch to concurrent fetching in the next step.

In this step, get ready to:
  • Write four synchronous fetch functions that each call a different public API.
  • Time the full sequential run and print how long it takes.
  • Format the aggregated API data into a readable report.
Define the API endpoints and fetch functions

Your aggregator needs to know which APIs to call. You will define four URL constants pointing at free, keyless APIs that return JSON data about weather, countries, users, and cryptocurrency prices.

  • In VS Code, open aggregator.py and replace the entire file contents with the following imports and URL constants:
import asyncio
import time

import httpx


# --- API URLs ---
WEATHER_URL = "https://api.open-meteo.com/v1/forecast?latitude=51.5&longitude=-0.1&current_weather=true"
COUNTRY_URL = "https://restcountries.com/v3.1/name/germany?fields=name,capital,population,flags"
USERS_URL = "https://randomuser.me/api/?results=3"
CRYPTO_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"

What do these URLs point to?

  • WEATHER_URL fetches the current weather for London from the Open-Meteo API.
  • COUNTRY_URL fetches Germany's name, capital, and population from REST Countries.
  • USERS_URL fetches 3 randomly generated user profiles from RandomUser.
  • CRYPTO_URL fetches live Bitcoin and Ethereum prices in USD from CoinGecko.

All four are free, require no API key, and return JSON. They also have different response times, which makes the sequential vs async comparison more interesting.

Now add the four synchronous fetch functions below the URL constants. Each one opens an httpx.Client connection, makes a GET request, checks for errors, and returns the parsed JSON.

  • Add this code directly below the URL constants in aggregator.py:
# --- Sequential Version ---
def fetch_weather_sync():
    with httpx.Client() as client:
        response = client.get(WEATHER_URL)
        response.raise_for_status()
        return response.json()


def fetch_country_sync():
    with httpx.Client() as client:
        response = client.get(COUNTRY_URL)
        response.raise_for_status()
        return response.json()


def fetch_users_sync():
    with httpx.Client() as client:
        response = client.get(USERS_URL)
        response.raise_for_status()
        return response.json()


def fetch_crypto_sync():
    with httpx.Client() as client:
        response = client.get(CRYPTO_URL)
        response.raise_for_status()
        return response.json()

What does this code do?

Each function follows the same three-step pattern:

  • with httpx.Client() as client: opens a connection pool that is automatically cleaned up when the block exits.
  • response.raise_for_status() raises an exception if the API returns an error status code (4xx or 5xx).
  • response.json() parses the response body from JSON into a Python dictionary.

These are standard synchronous HTTP calls. Each one blocks the entire program until it gets a response back, which is exactly the bottleneck you will fix later with async.

You won't see any output yet. The next chunk calls these functions and times the result.

Time the sequential fetch

Calling four APIs one after another means each request waits for the previous one to finish. The total time is the sum of all individual request times. You will wrap the calls with time.perf_counter() to measure this precisely.

  • Add the fetch_all_sequential() function below the four fetch functions in aggregator.py:
def fetch_all_sequential():
    start = time.perf_counter()
    weather = fetch_weather_sync()
    country = fetch_country_sync()
    users = fetch_users_sync()
    crypto = fetch_crypto_sync()
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"SEQUENTIAL: Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}

What does this code do?

  • time.perf_counter() captures a high-resolution timestamp before and after the API calls. The difference gives you the total elapsed time in seconds.
  • Each fetch function is called in sequence. The program is blocked during each request, so the total time equals the sum of all four response times.
  • The function returns a dictionary with all the data plus the elapsed time, so you can compare it to the async version later.

Now update main() to call this function so you can run the script and see the timing.

  • Find the existing main() function at the bottom of aggregator.py. It currently looks like this:
async def main():
    pass
  • Replace pass with the following two lines, so the function looks like this:
async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
  • Save aggregator.py.
  • Run the script by entering this command in your terminal:
python aggregator.py

You should see output like this (the exact time depends on your network speed):

Running sequential version...

==================================================
SEQUENTIAL: Fetched all 4 APIs in 2.34s
==================================================

What should I see?

The timing will typically be between 2 and 4 seconds. This is the number you will beat with async concurrency in the next step. If each API takes roughly 0.5 to 1 second, the sequential total is the sum of all four.

Seeing a connection error?

  • Make sure you have an active internet connection. All four APIs require network access.
  • Check that the URL constants are copied exactly. A missing character in the URL will cause a 404 or DNS error.
  • If you see httpx.ConnectError, your firewall or VPN may be blocking outbound HTTPS requests.

Why is my script throwing a connection error?

Display the aggregated results

The raw JSON from each API is hard to read. You will add a display_results() function that pulls out the most interesting fields and prints them in a clean, readable format.

  • Add the display_results() function below fetch_all_sequential() in aggregator.py:
# --- Display Results ---
def display_results(data: dict):
    print("\n--- AGGREGATED REPORT ---\n")

    if "weather" in data:
        weather = data["weather"]
        current = weather.get("current_weather", {})
        print(f"Weather in London: {current.get('temperature')}C, "
              f"Wind: {current.get('windspeed')} km/h")

    if "country" in data:
        country = data["country"]
        if isinstance(country, list) and len(country) > 0:
            c = country[0]
            print(f"Country: {c['name']['common']}, "
                  f"Capital: {c['capital'][0]}, "
                  f"Population: {c['population']:,}")

    if "users" in data:
        users = data["users"]
        user_list = users.get("results", [])
        names = [f"{u['name']['first']} {u['name']['last']}" for u in user_list]
        print(f"Random Users: {', '.join(names)}")

    if "crypto" in data:
        crypto = data["crypto"]
        btc = crypto.get("bitcoin", {}).get("usd", "N/A")
        eth = crypto.get("ethereum", {}).get("usd", "N/A")
        print(f"Crypto: BTC=${btc:,}, ETH=${eth:,}")

    print()

What does this code do?

The function checks which keys exist in the results dictionary, then extracts the most useful fields from each API's response:

  • Weather: current temperature and wind speed.
  • Country: name, capital city, and population (formatted with commas).
  • Users: first and last names of 3 random people.
  • Crypto: live USD prices for Bitcoin and Ethereum.

The if checks protect against missing data. If one API failed, the report still prints the others.

Now wire display_results() into main() so it prints the full report after fetching.

  • In aggregator.py, find the main() function. It currently looks like this:
async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
  • Add display_results(seq_results) on a new line below seq_results = fetch_all_sequential(), so the function looks like this:
async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)
  • Save aggregator.py and run it again:
python aggregator.py

You should now see both the timing banner and the formatted report with live data from all 4 APIs.

Report shows N/A or missing data?

  • If crypto prices show N/A, the CoinGecko API may be rate-limited. Wait 30 seconds and try again.
  • If user names are empty, check that USERS_URL includes ?results=3 at the end.
  • If you see a Python traceback, check that display_results is defined above main() in the file.

Why is my aggregated report showing missing data?

✔️ Awesome, I've got everything!

Great. Double check you have saved aggregator.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete aggregator.py at the end of this step:

import asyncio
import time

import httpx


# --- API URLs ---
WEATHER_URL = "https://api.open-meteo.com/v1/forecast?latitude=51.5&longitude=-0.1&current_weather=true"
COUNTRY_URL = "https://restcountries.com/v3.1/name/germany?fields=name,capital,population,flags"
USERS_URL = "https://randomuser.me/api/?results=3"
CRYPTO_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"


# --- Sequential Version ---
def fetch_weather_sync():
    with httpx.Client() as client:
        response = client.get(WEATHER_URL)
        response.raise_for_status()
        return response.json()


def fetch_country_sync():
    with httpx.Client() as client:
        response = client.get(COUNTRY_URL)
        response.raise_for_status()
        return response.json()


def fetch_users_sync():
    with httpx.Client() as client:
        response = client.get(USERS_URL)
        response.raise_for_status()
        return response.json()


def fetch_crypto_sync():
    with httpx.Client() as client:
        response = client.get(CRYPTO_URL)
        response.raise_for_status()
        return response.json()


def fetch_all_sequential():
    start = time.perf_counter()
    weather = fetch_weather_sync()
    country = fetch_country_sync()
    users = fetch_users_sync()
    crypto = fetch_crypto_sync()
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"SEQUENTIAL: Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Display Results ---
def display_results(data: dict):
    print("\n--- AGGREGATED REPORT ---\n")

    if "weather" in data:
        weather = data["weather"]
        current = weather.get("current_weather", {})
        print(f"Weather in London: {current.get('temperature')}C, "
              f"Wind: {current.get('windspeed')} km/h")

    if "country" in data:
        country = data["country"]
        if isinstance(country, list) and len(country) > 0:
            c = country[0]
            print(f"Country: {c['name']['common']}, "
                  f"Capital: {c['capital'][0]}, "
                  f"Population: {c['population']:,}")

    if "users" in data:
        users = data["users"]
        user_list = users.get("results", [])
        names = [f"{u['name']['first']} {u['name']['last']}" for u in user_list]
        print(f"Random Users: {', '.join(names)}")

    if "crypto" in data:
        crypto = data["crypto"]
        btc = crypto.get("bitcoin", {}).get("usd", "N/A")
        eth = crypto.get("ethereum", {}).get("usd", "N/A")
        print(f"Crypto: BTC=${btc:,}, ETH=${eth:,}")

    print()


async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)


if __name__ == "__main__":
    asyncio.run(main())

You now have hard evidence of how slow sequential API calls are. The total time is the sum of every individual request. Next up, you will convert these same calls to async coroutines and run them concurrently, cutting the total down to roughly the time of the single slowest request.

Convert to Async with asyncio.gather

In the previous step, your sequential fetcher called each API one after another. Each request had to finish completely before the next one could start.

The total wait time was the sum of all four requests. With asyncio.gather, you can fire all four requests at the same time and wait for them to finish together. The total time drops to roughly the slowest single request instead of the sum.

In this step, get ready to:
  • Rewrite your fetch functions as async coroutines using httpx.AsyncClient.
  • Run all four coroutines concurrently with asyncio.gather() and measure the speedup.
  • Add timeout protection so your aggregator fails gracefully if an API is slow.
Write the async fetch coroutines

Your sync functions each create their own httpx.Client connection. The async versions work differently.

Instead of each function managing its own connection, they'll all share a single httpx.AsyncClient passed in as a parameter. This lets httpx reuse the same underlying connections for better performance.

  • In aggregator.py, find the comment # --- Display Results --- and add the following code above it:
# --- Async Version with asyncio.gather ---
async def fetch_weather(client: httpx.AsyncClient):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    return response.json()


async def fetch_country(client: httpx.AsyncClient):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    return response.json()


async def fetch_users(client: httpx.AsyncClient):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    return response.json()


async def fetch_crypto(client: httpx.AsyncClient):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    return response.json()

What does this code do?

Each function is now a coroutine (defined with async def). The await keyword pauses the coroutine while the network request is in flight, freeing the event loop to run other coroutines during that wait.

The client: httpx.AsyncClient parameter means all four functions share one connection pool. You'll create the client once and pass it to all of them.

  • Save aggregator.py.

You won't see a change yet. These coroutines need a caller function to orchestrate them, which you'll write next.

Create the concurrent fetch function

asyncio.gather() takes multiple coroutines and schedules them all to run concurrently. It returns their results as a tuple in the same order you passed them in.

  • In aggregator.py, add the following function directly below fetch_crypto() (and still above the # --- Display Results --- comment):
async def fetch_all_gather():
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        weather, country, users, crypto = await asyncio.gather(
            fetch_weather(client),
            fetch_country(client),
            fetch_users(client),
            fetch_crypto(client),
        )
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (gather): Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}

What does this code do?

async with httpx.AsyncClient() as client: creates a shared async connection pool. The async with ensures the connection is properly closed when done.

asyncio.gather() fires all four coroutines at once. Python's event loop interleaves their network waits, so all four requests happen in parallel. The tuple unpacking weather, country, users, crypto = captures each result in order.

  • Now update main() to call your new async function. Find your existing main() function at the bottom of the file and replace it with:
async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)

    print("Running async (gather) version...")
    gather_results = await fetch_all_gather()
    display_results(gather_results)
  • Save aggregator.py and run it:
python aggregator.py

You should see both versions run, each printing their timing. The async version should complete in roughly 0.3-0.8 seconds compared to 2-4 seconds for the sequential version.

Not seeing a speedup?

Make sure your async functions use await client.get() and not just client.get(). Without await, the request won't actually execute.

Also confirm your coroutines use async def (not plain def). A regular function passed to asyncio.gather() will raise a TypeError.

Help me debug my asyncio.gather timing

Add timeout protection

In production, APIs can hang indefinitely. asyncio.timeout() wraps an async operation with a deadline. If the operation takes longer than the specified seconds, Python raises a TimeoutError that you can catch and handle gracefully.

  • In aggregator.py, find your fetch_all_gather() function and replace the entire function with this updated version:
async def fetch_all_gather():
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.timeout(10):
                weather, country, users, crypto = await asyncio.gather(
                    fetch_weather(client),
                    fetch_country(client),
                    fetch_users(client),
                    fetch_crypto(client),
                )
        except TimeoutError:
            print("ERROR: Requests timed out after 10 seconds!")
            return None
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (gather): Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}

What changed here?

async with asyncio.timeout(10): creates a 10-second deadline around the gather call. If any of the requests push the total time past 10 seconds, Python cancels all pending tasks and raises TimeoutError.

The except TimeoutError: block catches the error and returns None so the rest of your program can continue instead of crashing.

  • Since fetch_all_gather() can now return None on timeout, update main() to handle that. Find the line display_results(gather_results) and replace it with:
    if gather_results:
        display_results(gather_results)
  • Save aggregator.py and run it again:
python aggregator.py

You should see the same timing comparison as before. The timeout won't trigger under normal conditions, but your aggregator is now protected against hanging APIs.

Seeing a NameError for TimeoutError?

In Python 3.11+, TimeoutError is a built-in exception. You do not need to import it from asyncio. If you're seeing an error, make sure you wrote except TimeoutError: (not except asyncio.TimeoutError:).

Help me fix my timeout error handling

✔️ Awesome, I've got everything!

Great. Double check you've saved aggregator.py and your script runs without errors.

ⓧ I'd like to double check the full code

import asyncio
import time

import httpx


# --- API URLs ---
WEATHER_URL = "https://api.open-meteo.com/v1/forecast?latitude=51.5&longitude=-0.1&current_weather=true"
COUNTRY_URL = "https://restcountries.com/v3.1/name/germany?fields=name,capital,population,flags"
USERS_URL = "https://randomuser.me/api/?results=3"
CRYPTO_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"


# --- Sequential Version ---
def fetch_weather_sync():
    with httpx.Client() as client:
        response = client.get(WEATHER_URL)
        response.raise_for_status()
        return response.json()


def fetch_country_sync():
    with httpx.Client() as client:
        response = client.get(COUNTRY_URL)
        response.raise_for_status()
        return response.json()


def fetch_users_sync():
    with httpx.Client() as client:
        response = client.get(USERS_URL)
        response.raise_for_status()
        return response.json()


def fetch_crypto_sync():
    with httpx.Client() as client:
        response = client.get(CRYPTO_URL)
        response.raise_for_status()
        return response.json()


def fetch_all_sequential():
    start = time.perf_counter()
    weather = fetch_weather_sync()
    country = fetch_country_sync()
    users = fetch_users_sync()
    crypto = fetch_crypto_sync()
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"SEQUENTIAL: Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Async Version with asyncio.gather ---
async def fetch_weather(client: httpx.AsyncClient):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    return response.json()


async def fetch_country(client: httpx.AsyncClient):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    return response.json()


async def fetch_users(client: httpx.AsyncClient):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    return response.json()


async def fetch_crypto(client: httpx.AsyncClient):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    return response.json()


async def fetch_all_gather():
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.timeout(10):
                weather, country, users, crypto = await asyncio.gather(
                    fetch_weather(client),
                    fetch_country(client),
                    fetch_users(client),
                    fetch_crypto(client),
                )
        except TimeoutError:
            print("ERROR: Requests timed out after 10 seconds!")
            return None
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (gather): Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Display Results ---
def display_results(data: dict):
    print("\n--- AGGREGATED REPORT ---\n")

    if "weather" in data:
        weather = data["weather"]
        current = weather.get("current_weather", {})
        print(f"Weather in London: {current.get('temperature')}C, "
              f"Wind: {current.get('windspeed')} km/h")

    if "country" in data:
        country = data["country"]
        if isinstance(country, list) and len(country) > 0:
            c = country[0]
            print(f"Country: {c['name']['common']}, "
                  f"Capital: {c['capital'][0]}, "
                  f"Population: {c['population']:,}")

    if "users" in data:
        users = data["users"]
        user_list = users.get("results", [])
        names = [f"{u['name']['first']} {u['name']['last']}" for u in user_list]
        print(f"Random Users: {', '.join(names)}")

    if "crypto" in data:
        crypto = data["crypto"]
        btc = crypto.get("bitcoin", {}).get("usd", "N/A")
        eth = crypto.get("ethereum", {}).get("usd", "N/A")
        print(f"Crypto: BTC=${btc:,}, ETH=${eth:,}")

    print()


async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)

    print("Running async (gather) version...")
    gather_results = await fetch_all_gather()
    if gather_results:
        display_results(gather_results)


if __name__ == "__main__":
    asyncio.run(main())

Your aggregator now fetches data from all four APIs concurrently with timeout protection. Next up, you'll upgrade to TaskGroup for structured concurrency that automatically cancels sibling tasks when one fails.

Upgrade to TaskGroup

Your asyncio.gather approach already proves that concurrent fetching is faster. But it has a blind spot: if one task crashes, the others keep running without knowing something went wrong.

Python's TaskGroup solves this with structured concurrency. If any task raises an exception, TaskGroup cancels all remaining tasks and collects the errors into an ExceptionGroup. In this step, you will rewrite the aggregator using TaskGroup, add deliberate error handling, and use Python 3.14's new call graph introspection to visualize your task hierarchy.

In this step, get ready to:
  • Rewrite the concurrent fetcher using TaskGroup and structured concurrency.
  • Add a broken API call and handle failures gracefully with ExceptionGroup.
  • Use Python 3.14's call graph introspection to visualize the async task tree.
Rewrite the fetch functions for TaskGroup

TaskGroup tasks cannot return values directly the way asyncio.gather does. Instead, each task writes its result into a shared dictionary. This means you need slightly different versions of your fetch coroutines.

  • In aggregator.py, find the display_results function. Add the following code above it to create four task-style async functions:
# --- Async Version with TaskGroup ---
async def fetch_weather_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["weather"] = response.json()


async def fetch_country_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    results["country"] = response.json()


async def fetch_users_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    results["users"] = response.json()


async def fetch_crypto_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    results["crypto"] = response.json()

What does this code do?

Each function takes both an httpx.AsyncClient and a results dictionary. Instead of returning a value, it stores the parsed JSON directly into the shared dict under a named key.

This write-to-dict pattern is how TaskGroup tasks communicate results back to the caller. The parent function creates an empty results = {} dict, spawns all tasks, and reads the dict after the TaskGroup block exits.

  • Now add the fetch_all_taskgroup() function directly below the four task functions you just added:
async def fetch_all_taskgroup():
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_weather_task(client, results))
                tg.create_task(fetch_country_task(client, results))
                tg.create_task(fetch_users_task(client, results))
                tg.create_task(fetch_crypto_task(client, results))
        except ExceptionGroup as eg:
            print(f"\nTaskGroup caught {len(eg.exceptions)} error(s):")
            for i, exc in enumerate(eg.exceptions, 1):
                print(f"  {i}. {type(exc).__name__}: {exc}")
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (TaskGroup): Fetched APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    results["time"] = elapsed
    return results

What does this code do?

The async with asyncio.TaskGroup() as tg: block spawns all four tasks concurrently using tg.create_task(). The block waits until every task finishes before continuing.

If any task raises an exception, TaskGroup immediately cancels all remaining tasks and wraps the errors into an ExceptionGroup. The except ExceptionGroup as eg: handler catches this and prints each individual error.

Why TaskGroup instead of gather?

With asyncio.gather, if one task fails the others keep running as zombie tasks you never hear from again. This leaks resources and makes debugging harder.

TaskGroup enforces structured concurrency. Every task you spawn is guaranteed to complete (or be cancelled) before the async with block exits. No orphans, no surprises.

  • Save aggregator.py.

Seeing a NameError for ExceptionGroup?

ExceptionGroup is a built-in in Python 3.11+. If you see a NameError, confirm you are running Python 3.14 by checking python --version in your terminal. Make sure your virtual environment was created with your Python 3.14 installation.

help me fix a NameError for ExceptionGroup in my async aggregator

Add a broken API to demonstrate error handling

The real power of TaskGroup shows up when things go wrong. To see automatic cancellation in action, you will add a fifth API call that always fails and watch how TaskGroup handles it gracefully.

  • In aggregator.py, add this broken API function directly below your fetch_crypto_task function (above fetch_all_taskgroup):
async def fetch_broken_api(client: httpx.AsyncClient, results: dict):
    """A deliberately broken API call to demonstrate TaskGroup error handling."""
    response = await client.get("https://httpbin.org/status/500")
    response.raise_for_status()
    results["broken"] = response.json()

What does this code do?

This function requests a URL that always returns HTTP 500. When raise_for_status() runs, it throws an httpx.HTTPStatusError. Inside a TaskGroup, this exception propagates up and cancels all sibling tasks.

  • Now add a second orchestrator function that includes the broken task. Place this directly below fetch_all_taskgroup():
# --- TaskGroup with Error Handling Demo ---
async def fetch_all_taskgroup_with_errors():
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_weather_task(client, results))
                tg.create_task(fetch_country_task(client, results))
                tg.create_task(fetch_users_task(client, results))
                tg.create_task(fetch_crypto_task(client, results))
                tg.create_task(fetch_broken_api(client, results))
        except ExceptionGroup as eg:
            print(f"\nTaskGroup caught {len(eg.exceptions)} error(s):")
            for i, exc in enumerate(eg.exceptions, 1):
                print(f"  {i}. {type(exc).__name__}: {exc}")
            print("Continuing with partial results...")
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (TaskGroup + error handling): Completed in {elapsed:.2f}s")
    print(f"{'='*50}")
    results["time"] = elapsed
    return results

What does this code do?

This function spawns the same four working tasks plus the broken one. When fetch_broken_api fails, TaskGroup cancels any tasks that haven't finished yet and collects all errors.

The except ExceptionGroup handler iterates over eg.exceptions to print each failure individually. Because you catch the group, the script continues with whatever partial results the successful tasks already wrote into the dict.

  • Save aggregator.py.

Seeing an indentation error?

Make sure the tg.create_task(...) lines are indented inside both the try: block and the async with asyncio.TaskGroup() block. That is three levels of indentation from the function definition.

help me fix indentation in my TaskGroup function

Visualize the task tree with call graph introspection

Python 3.14 added asyncio.print_call_graph() to help you see which tasks are running and which parent task spawned them. This is invaluable for debugging complex async programs where you lose track of task ownership.

  • Add the introspection functions below fetch_all_taskgroup_with_errors() and above display_results():
# --- Call Graph Introspection Demo ---
async def fetch_weather_with_introspection(client: httpx.AsyncClient, results: dict):
    asyncio.print_call_graph()
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["weather"] = response.json()


async def introspection_demo():
    print("\n--- Call Graph Introspection ---")
    results = {}
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(
                fetch_weather_with_introspection(client, results),
                name="weather-task",
            )
            tg.create_task(fetch_country_task(client, results), name="country-task")
            tg.create_task(fetch_users_task(client, results), name="users-task")
            tg.create_task(fetch_crypto_task(client, results), name="crypto-task")
    print("--- End Introspection ---\n")

What does this code do?

The fetch_weather_with_introspection function calls asyncio.print_call_graph() before making its HTTP request. This prints the current task's identity, its call stack, and which parent task is awaiting it.

The introspection_demo function creates named tasks using the name parameter on tg.create_task(). Named tasks make the call graph output much easier to read.

  • Now replace your entire main() function with this updated version that runs all three approaches plus the introspection demo:
async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)

    print("Running async (gather) version...")
    gather_results = await fetch_all_gather()
    if gather_results:
        display_results(gather_results)

    print("Running async (TaskGroup) version...")
    tg_results = await fetch_all_taskgroup()
    display_results(tg_results)

    print("Running async (TaskGroup + broken API) version...")
    error_results = await fetch_all_taskgroup_with_errors()
    display_results(error_results)

    await introspection_demo()

What does the updated main do?

The updated main() runs four demos in sequence: the original sequential version, the gather version, the TaskGroup version, and the TaskGroup version with the broken API included. It finishes by running the introspection demo so you can see the call graph output.

  • Save aggregator.py.
  • Run the script by executing this command:
python aggregator.py

You should see the sequential and gather output first, then the TaskGroup timing, followed by the error handling demo printing caught errors, and finally the call graph introspection output.

The call graph output shows your task name, its call stack, and an "Awaited by" section pointing to the parent task that spawned it.

Not seeing the call graph output?

Make sure you are running Python 3.14 or newer. The asyncio.print_call_graph() function was added in Python 3.14 and does not exist in earlier versions. Run python --version to confirm.

help me troubleshoot missing call graph output in my async aggregator

✔️ Awesome, I've got everything!

Great. Double check you have saved aggregator.py with all the new functions.

ⓧ I'd like to double check the full code

import asyncio
import time

import httpx


# --- API URLs ---
WEATHER_URL = "https://api.open-meteo.com/v1/forecast?latitude=51.5&longitude=-0.1&current_weather=true"
COUNTRY_URL = "https://restcountries.com/v3.1/name/germany?fields=name,capital,population,flags"
USERS_URL = "https://randomuser.me/api/?results=3"
CRYPTO_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"


# --- Sequential Version ---
def fetch_weather_sync():
    with httpx.Client() as client:
        response = client.get(WEATHER_URL)
        response.raise_for_status()
        return response.json()


def fetch_country_sync():
    with httpx.Client() as client:
        response = client.get(COUNTRY_URL)
        response.raise_for_status()
        return response.json()


def fetch_users_sync():
    with httpx.Client() as client:
        response = client.get(USERS_URL)
        response.raise_for_status()
        return response.json()


def fetch_crypto_sync():
    with httpx.Client() as client:
        response = client.get(CRYPTO_URL)
        response.raise_for_status()
        return response.json()


def fetch_all_sequential():
    start = time.perf_counter()
    weather = fetch_weather_sync()
    country = fetch_country_sync()
    users = fetch_users_sync()
    crypto = fetch_crypto_sync()
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"SEQUENTIAL: Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Async Version with asyncio.gather ---
async def fetch_weather(client: httpx.AsyncClient):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    return response.json()


async def fetch_country(client: httpx.AsyncClient):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    return response.json()


async def fetch_users(client: httpx.AsyncClient):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    return response.json()


async def fetch_crypto(client: httpx.AsyncClient):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    return response.json()


async def fetch_all_gather():
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.timeout(10):
                weather, country, users, crypto = await asyncio.gather(
                    fetch_weather(client),
                    fetch_country(client),
                    fetch_users(client),
                    fetch_crypto(client),
                )
        except TimeoutError:
            print("ERROR: Requests timed out after 10 seconds!")
            return None
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (gather): Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Async Version with TaskGroup ---
async def fetch_weather_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["weather"] = response.json()


async def fetch_country_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    results["country"] = response.json()


async def fetch_users_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    results["users"] = response.json()


async def fetch_crypto_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    results["crypto"] = response.json()


async def fetch_broken_api(client: httpx.AsyncClient, results: dict):
    """A deliberately broken API call to demonstrate TaskGroup error handling."""
    response = await client.get("https://httpbin.org/status/500")
    response.raise_for_status()
    results["broken"] = response.json()


async def fetch_all_taskgroup():
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_weather_task(client, results))
                tg.create_task(fetch_country_task(client, results))
                tg.create_task(fetch_users_task(client, results))
                tg.create_task(fetch_crypto_task(client, results))
        except ExceptionGroup as eg:
            print(f"\nTaskGroup caught {len(eg.exceptions)} error(s):")
            for i, exc in enumerate(eg.exceptions, 1):
                print(f"  {i}. {type(exc).__name__}: {exc}")
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (TaskGroup): Fetched APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    results["time"] = elapsed
    return results


# --- TaskGroup with Error Handling Demo ---
async def fetch_all_taskgroup_with_errors():
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_weather_task(client, results))
                tg.create_task(fetch_country_task(client, results))
                tg.create_task(fetch_users_task(client, results))
                tg.create_task(fetch_crypto_task(client, results))
                tg.create_task(fetch_broken_api(client, results))
        except ExceptionGroup as eg:
            print(f"\nTaskGroup caught {len(eg.exceptions)} error(s):")
            for i, exc in enumerate(eg.exceptions, 1):
                print(f"  {i}. {type(exc).__name__}: {exc}")
            print("Continuing with partial results...")
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (TaskGroup + error handling): Completed in {elapsed:.2f}s")
    print(f"{'='*50}")
    results["time"] = elapsed
    return results


# --- Call Graph Introspection Demo ---
async def fetch_weather_with_introspection(client: httpx.AsyncClient, results: dict):
    asyncio.print_call_graph()
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["weather"] = response.json()


async def introspection_demo():
    print("\n--- Call Graph Introspection ---")
    results = {}
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(
                fetch_weather_with_introspection(client, results),
                name="weather-task",
            )
            tg.create_task(fetch_country_task(client, results), name="country-task")
            tg.create_task(fetch_users_task(client, results), name="users-task")
            tg.create_task(fetch_crypto_task(client, results), name="crypto-task")
    print("--- End Introspection ---\n")


# --- Display Results ---
def display_results(data: dict):
    print("\n--- AGGREGATED REPORT ---\n")

    if "weather" in data:
        weather = data["weather"]
        current = weather.get("current_weather", {})
        print(f"Weather in London: {current.get('temperature')}C, "
              f"Wind: {current.get('windspeed')} km/h")

    if "country" in data:
        country = data["country"]
        if isinstance(country, list) and len(country) > 0:
            c = country[0]
            print(f"Country: {c['name']['common']}, "
                  f"Capital: {c['capital'][0]}, "
                  f"Population: {c['population']:,}")

    if "users" in data:
        users = data["users"]
        user_list = users.get("results", [])
        names = [f"{u['name']['first']} {u['name']['last']}" for u in user_list]
        print(f"Random Users: {', '.join(names)}")

    if "crypto" in data:
        crypto = data["crypto"]
        btc = crypto.get("bitcoin", {}).get("usd", "N/A")
        eth = crypto.get("ethereum", {}).get("usd", "N/A")
        print(f"Crypto: BTC=${btc:,}, ETH=${eth:,}")

    print()


async def main():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)

    print("Running async (gather) version...")
    gather_results = await fetch_all_gather()
    if gather_results:
        display_results(gather_results)

    print("Running async (TaskGroup) version...")
    tg_results = await fetch_all_taskgroup()
    display_results(tg_results)

    print("Running async (TaskGroup + broken API) version...")
    error_results = await fetch_all_taskgroup_with_errors()
    display_results(error_results)

    await introspection_demo()


if __name__ == "__main__":
    asyncio.run(main())

You now have structured concurrency protecting your aggregator from zombie tasks. Next up, you will put all three approaches head-to-head with a timing comparison and simulate slow APIs to see how async shines under load.

Compare and Analyze Results

You've built sequential, gather, and TaskGroup versions of your aggregator. Each one printed its own timing, but there's no single summary that puts them side by side.

In this step, you'll create a dedicated comparison function that collects all timings into a formatted summary table. Then you'll simulate a slow API to prove that async handles delays without blocking other tasks.

In this step, get ready to:
  • Create a comparison function with a performance summary table.
  • Simulate a slow API call and observe how async handles it.
Create the comparison function

Right now your main() function runs every version and prints results, but the timing data disappears into the scroll. A dedicated comparison function can collect all the timings and print them in a clean table at the end.

  • In aggregator.py, find async def main():. Add the following compare() function directly ABOVE it, between display_results() and main():
# --- Comparison ---
async def compare():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)

    print("Running async (gather) version...")
    gather_results = await fetch_all_gather()
    if gather_results:
        display_results(gather_results)

    print("Running async (TaskGroup) version...")
    tg_results = await fetch_all_taskgroup()
    display_results(tg_results)

    print("Running async (TaskGroup + broken API) version...")
    error_results = await fetch_all_taskgroup_with_errors()
    display_results(error_results)

    await introspection_demo()

    # Summary
    print("\n" + "=" * 50)
    print("PERFORMANCE COMPARISON")
    print("=" * 50)
    seq_time = seq_results["time"]
    gather_time = gather_results["time"] if gather_results else 0
    tg_time = tg_results["time"]
    print(f"  Sequential:     {seq_time:.2f}s")
    print(f"  asyncio.gather: {gather_time:.2f}s")
    print(f"  TaskGroup:      {tg_time:.2f}s")
    if gather_time > 0:
        print(f"  Speedup:        {seq_time / gather_time:.1f}x faster with async")
    print("=" * 50)

What does this code do?

The first half of compare() is the same logic your main() already had. It runs the sequential version, the gather version, the TaskGroup version, the error-handling demo, and the introspection demo.

The new part is the summary table at the end. It extracts the "time" key from each result dictionary, prints them in a formatted table, and calculates a speedup factor by dividing the sequential time by the gather time.

  • Now replace your existing main() function with this simplified version that just calls compare():
async def main():
    await compare()


if __name__ == "__main__":
    asyncio.run(main())
  • Save aggregator.py.
  • Run the script by running:
python aggregator.py
  • At the bottom of the output, you should see a PERFORMANCE COMPARISON table showing times for all three approaches and a speedup factor (typically 3-5x).

Not seeing the comparison table?

If you see a NameError for compare, make sure the compare() function is defined above main() in the file. Python reads top to bottom, so functions must be defined before they're called.

If the table prints but shows 0.00s for gather, check that fetch_all_gather() is not returning None due to a timeout. Your network may be slow. Try increasing the timeout value in that function.

Help me debug my comparison function.

Simulate a slow API call

The comparison table shows real numbers, but network conditions vary between runs. To prove async's value consistently, you can simulate a slow API with a fixed 2-second delay and compare how it affects sequential vs concurrent total time.

  • In aggregator.py, add the following code between the compare() function and main():
# --- Simulated Slow API Demo ---
async def fetch_slow_api(client: httpx.AsyncClient, results: dict):
    """Simulate a slow API by adding a 2-second delay."""
    await asyncio.sleep(2)
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["slow_weather"] = response.json()


async def slow_demo():
    print("\n--- Slow API Demo ---")
    print("Adding a 2-second simulated delay to one API call...\n")

    # Sequential with delay
    start = time.perf_counter()
    with httpx.Client() as client:
        client.get(WEATHER_URL)  # normal
        time.sleep(2)  # simulate slow
        client.get(COUNTRY_URL)  # normal
    seq_time = time.perf_counter() - start
    print(f"  Sequential (with 2s slow call): {seq_time:.2f}s")

    # Async with delay
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_slow_api(client, results))
            tg.create_task(fetch_country_task(client, results))
            tg.create_task(fetch_users_task(client, results))
            tg.create_task(fetch_crypto_task(client, results))
    async_time = time.perf_counter() - start
    print(f"  Async TaskGroup (with 2s slow call): {async_time:.2f}s")
    print(f"  The slow call didn't block others! Total ~ max(2s, other calls)")
    print("--- End Slow Demo ---\n")

What does this code do?

The fetch_slow_api() function uses await asyncio.sleep(2) to simulate a 2-second network delay before making its real request. Because it uses asyncio.sleep() instead of time.sleep(), the delay is non-blocking. Other tasks run during the wait.

The slow_demo() function runs the same scenario both ways. The sequential version uses blocking time.sleep(2) which freezes everything for 2 seconds. The async version runs the slow task alongside the other three tasks concurrently, so the total time is roughly the maximum of all tasks (about 2 seconds) rather than the sum.

  • In aggregator.py, find this line inside main():
async def main():
    await compare()
  • Add await slow_demo() below await compare(), so the function looks like this:
async def main():
    await compare()
    await slow_demo()
  • Save aggregator.py and run it:
python aggregator.py
  • At the very end of the output, you should see the Slow API Demo section. The sequential version with the 2-second delay should take around 4 seconds total, while the async version should take only around 2 seconds because the other tasks run during the sleep.

Why does the async version still take 2 seconds?

The async version's total time is the maximum of all concurrent task durations, not the sum. Since fetch_slow_api() sleeps for 2 seconds and the other API calls each take less than 1 second, the group finishes in roughly 2 seconds. The key insight is that the other three tasks complete during that 2-second sleep instead of waiting for it.

Sequential and async times look the same?

Make sure fetch_slow_api() uses await asyncio.sleep(2) (non-blocking) and that the sequential section uses time.sleep(2) (blocking). If you accidentally use time.sleep() in the async function, it blocks the entire event loop and defeats the purpose of concurrency.

Help me understand why my async slow demo isn't faster than sequential.

✔️ Awesome, I've got everything!

Double check you've saved your file and your terminal shows both the performance comparison table and the slow API demo output.

ⓧ I'd like to double check the full code

import asyncio
import time

import httpx


# --- API URLs ---
WEATHER_URL = "https://api.open-meteo.com/v1/forecast?latitude=51.5&longitude=-0.1&current_weather=true"
COUNTRY_URL = "https://restcountries.com/v3.1/name/germany?fields=name,capital,population,flags"
USERS_URL = "https://randomuser.me/api/?results=3"
CRYPTO_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"


# --- Sequential Version ---
def fetch_weather_sync():
    with httpx.Client() as client:
        response = client.get(WEATHER_URL)
        response.raise_for_status()
        return response.json()


def fetch_country_sync():
    with httpx.Client() as client:
        response = client.get(COUNTRY_URL)
        response.raise_for_status()
        return response.json()


def fetch_users_sync():
    with httpx.Client() as client:
        response = client.get(USERS_URL)
        response.raise_for_status()
        return response.json()


def fetch_crypto_sync():
    with httpx.Client() as client:
        response = client.get(CRYPTO_URL)
        response.raise_for_status()
        return response.json()


def fetch_all_sequential():
    start = time.perf_counter()
    weather = fetch_weather_sync()
    country = fetch_country_sync()
    users = fetch_users_sync()
    crypto = fetch_crypto_sync()
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"SEQUENTIAL: Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Async Version with asyncio.gather ---
async def fetch_weather(client: httpx.AsyncClient):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    return response.json()


async def fetch_country(client: httpx.AsyncClient):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    return response.json()


async def fetch_users(client: httpx.AsyncClient):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    return response.json()


async def fetch_crypto(client: httpx.AsyncClient):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    return response.json()


async def fetch_all_gather():
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.timeout(10):
                weather, country, users, crypto = await asyncio.gather(
                    fetch_weather(client),
                    fetch_country(client),
                    fetch_users(client),
                    fetch_crypto(client),
                )
        except TimeoutError:
            print("ERROR: Requests timed out after 10 seconds!")
            return None
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (gather): Fetched all 4 APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    return {"weather": weather, "country": country, "users": users, "crypto": crypto, "time": elapsed}


# --- Async Version with TaskGroup ---
async def fetch_weather_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["weather"] = response.json()


async def fetch_country_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(COUNTRY_URL)
    response.raise_for_status()
    results["country"] = response.json()


async def fetch_users_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(USERS_URL)
    response.raise_for_status()
    results["users"] = response.json()


async def fetch_crypto_task(client: httpx.AsyncClient, results: dict):
    response = await client.get(CRYPTO_URL)
    response.raise_for_status()
    results["crypto"] = response.json()


async def fetch_broken_api(client: httpx.AsyncClient, results: dict):
    """A deliberately broken API call to demonstrate TaskGroup error handling."""
    response = await client.get("https://httpbin.org/status/500")
    response.raise_for_status()
    results["broken"] = response.json()


async def fetch_all_taskgroup():
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_weather_task(client, results))
                tg.create_task(fetch_country_task(client, results))
                tg.create_task(fetch_users_task(client, results))
                tg.create_task(fetch_crypto_task(client, results))
        except ExceptionGroup as eg:
            print(f"\nTaskGroup caught {len(eg.exceptions)} error(s):")
            for i, exc in enumerate(eg.exceptions, 1):
                print(f"  {i}. {type(exc).__name__}: {exc}")
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (TaskGroup): Fetched APIs in {elapsed:.2f}s")
    print(f"{'='*50}")
    results["time"] = elapsed
    return results


async def fetch_all_taskgroup_with_errors():
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_weather_task(client, results))
                tg.create_task(fetch_country_task(client, results))
                tg.create_task(fetch_users_task(client, results))
                tg.create_task(fetch_crypto_task(client, results))
                tg.create_task(fetch_broken_api(client, results))
        except ExceptionGroup as eg:
            print(f"\nTaskGroup caught {len(eg.exceptions)} error(s):")
            for i, exc in enumerate(eg.exceptions, 1):
                print(f"  {i}. {type(exc).__name__}: {exc}")
            print("Continuing with partial results...")
    elapsed = time.perf_counter() - start
    print(f"\n{'='*50}")
    print(f"ASYNC (TaskGroup + error handling): Completed in {elapsed:.2f}s")
    print(f"{'='*50}")
    results["time"] = elapsed
    return results


# --- Call Graph Introspection Demo ---
async def fetch_weather_with_introspection(client: httpx.AsyncClient, results: dict):
    asyncio.print_call_graph()
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["weather"] = response.json()


async def introspection_demo():
    print("\n--- Call Graph Introspection ---")
    results = {}
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(
                fetch_weather_with_introspection(client, results),
                name="weather-task",
            )
            tg.create_task(fetch_country_task(client, results), name="country-task")
            tg.create_task(fetch_users_task(client, results), name="users-task")
            tg.create_task(fetch_crypto_task(client, results), name="crypto-task")
    print("--- End Introspection ---\n")


# --- Display Results ---
def display_results(data: dict):
    print("\n--- AGGREGATED REPORT ---\n")

    if "weather" in data:
        weather = data["weather"]
        current = weather.get("current_weather", {})
        print(f"Weather in London: {current.get('temperature')}C, "
              f"Wind: {current.get('windspeed')} km/h")

    if "country" in data:
        country = data["country"]
        if isinstance(country, list) and len(country) > 0:
            c = country[0]
            print(f"Country: {c['name']['common']}, "
                  f"Capital: {c['capital'][0]}, "
                  f"Population: {c['population']:,}")

    if "users" in data:
        users = data["users"]
        user_list = users.get("results", [])
        names = [f"{u['name']['first']} {u['name']['last']}" for u in user_list]
        print(f"Random Users: {', '.join(names)}")

    if "crypto" in data:
        crypto = data["crypto"]
        btc = crypto.get("bitcoin", {}).get("usd", "N/A")
        eth = crypto.get("ethereum", {}).get("usd", "N/A")
        print(f"Crypto: BTC=${btc:,}, ETH=${eth:,}")

    print()


# --- Comparison ---
async def compare():
    print("Running sequential version...")
    seq_results = fetch_all_sequential()
    display_results(seq_results)

    print("Running async (gather) version...")
    gather_results = await fetch_all_gather()
    if gather_results:
        display_results(gather_results)

    print("Running async (TaskGroup) version...")
    tg_results = await fetch_all_taskgroup()
    display_results(tg_results)

    print("Running async (TaskGroup + broken API) version...")
    error_results = await fetch_all_taskgroup_with_errors()
    display_results(error_results)

    await introspection_demo()

    # Summary
    print("\n" + "=" * 50)
    print("PERFORMANCE COMPARISON")
    print("=" * 50)
    seq_time = seq_results["time"]
    gather_time = gather_results["time"] if gather_results else 0
    tg_time = tg_results["time"]
    print(f"  Sequential:     {seq_time:.2f}s")
    print(f"  asyncio.gather: {gather_time:.2f}s")
    print(f"  TaskGroup:      {tg_time:.2f}s")
    if gather_time > 0:
        print(f"  Speedup:        {seq_time / gather_time:.1f}x faster with async")
    print("=" * 50)


# --- Simulated Slow API Demo ---
async def fetch_slow_api(client: httpx.AsyncClient, results: dict):
    """Simulate a slow API by adding a 2-second delay."""
    await asyncio.sleep(2)
    response = await client.get(WEATHER_URL)
    response.raise_for_status()
    results["slow_weather"] = response.json()


async def slow_demo():
    print("\n--- Slow API Demo ---")
    print("Adding a 2-second simulated delay to one API call...\n")

    # Sequential with delay
    start = time.perf_counter()
    with httpx.Client() as client:
        client.get(WEATHER_URL)  # normal
        time.sleep(2)  # simulate slow
        client.get(COUNTRY_URL)  # normal
    seq_time = time.perf_counter() - start
    print(f"  Sequential (with 2s slow call): {seq_time:.2f}s")

    # Async with delay
    start = time.perf_counter()
    results = {}
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_slow_api(client, results))
            tg.create_task(fetch_country_task(client, results))
            tg.create_task(fetch_users_task(client, results))
            tg.create_task(fetch_crypto_task(client, results))
    async_time = time.perf_counter() - start
    print(f"  Async TaskGroup (with 2s slow call): {async_time:.2f}s")
    print(f"  The slow call didn't block others! Total ~ max(2s, other calls)")
    print("--- End Slow Demo ---\n")


async def main():
    await compare()
    await slow_demo()


if __name__ == "__main__":
    asyncio.run(main())

Secret mission

Add Retry Logic with Exponential Backoff

Your async aggregator is fast, but what happens when an API goes down? In production, APIs fail all the time. Your mission: add retry logic with exponential backoff and a circuit breaker pattern that gracefully handles unreliable endpoints without crashing the whole aggregator.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The async-aggregator project folder on your Desktop (contains aggregator.py, the venv virtual environment, and installed packages).

Keep everything running

No action needed. Choose this if you're still actively building or want to experiment further with the aggregator script.

  • Your project folder stays on your Desktop, ready to use anytime.
  • To resume work later, open VS Code, navigate to async-aggregator, and activate the virtual environment again.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files stay on disk for next time.

  • Deactivate your virtual environment by running:
deactivate
  • Close the VS Code window for this project.
  • To resume later, reopen the folder in VS Code, open a terminal, and run venv\Scripts\activate to reactivate your environment.

Delete - I don't want to use this again

Remove all project files and start fresh. This deletes your code, virtual environment, and all installed packages.

  • Deactivate your virtual environment first by running:
deactivate
  • Close VS Code, then delete the project folder.

On macOS or Linux:

rm -rf ~/Desktop/async-aggregator

On Windows:

rmdir /s /q %USERPROFILE%\Desktop\async-aggregator
  • Confirm the folder is gone by checking your Desktop. You should no longer see async-aggregator listed.

Nice Work!

Nice Work!

You just built a production-ready async API aggregator that fetches data from 4 live APIs concurrently using Python asyncio.

You've learned how to:

  • Orchestrated concurrent API requests with asyncio.gather() and asyncio.TaskGroup, proving a measurable speedup over sequential fetching.
  • Handled structured concurrency errors using ExceptionGroup, timeouts, and automatic task cancellation when sibling tasks fail.
  • Visualized your async task tree using Python 3.14's print_call_graph() introspection tools to debug task hierarchies.
  • Secret Mission: Added retry logic with exponential backoff and a circuit breaker pattern for resilient API calls that degrade gracefully under failure.

Ready to quiz yourself?