python asyncio async await coroutines event loop 2024 Challenge

Advanced Coding Challenge: The High-Performance Asynchronous Data Aggregator

Problem Description Imagine you're actually building a blazing-fast data aggregation dashboard to a startup. Your task is to fetch live user statistics from a list of third-party analytics API endpoints.

Here is the massive real-world problem: you must use a standard blocking requests library (known as "HTTP for Humans"), but if you call requests.get() sequentially for 100 URLs, your Python program will completely freeze and stop doing anything else until each server replies.

To prevent your program from freezing you've got to implement Asynchronous I/O. Your challenge is to bridge standard blocking code with modern async code by converting your blocking HTTP requests into coroutines using asyncio.to_thread().

Plus a third-party API is simply notoriously unreliable. It frequently times out, occasionally returns flat HTML instead of formatted JSON and actively blocks default Python scripts to stop spam bots. Your solution must practice safe deserialization handle custom metadata envelopes (headers), and gracefully return None when a server fails rather than crashing your entire application.

Difficulty Level: Advanced

Input & Output Specifications * Input: * api_urls (List of strings): A list of endpoint URLs for fetch data from. * api_token (string): A secret token needed to prove you have permission to access data. * Output: * Returns List containing results for each URL. * Each successful result should really be the parsed Python Dictionary (the deserialized JSON). * If a request fails due to a network error, timeout or invalid JSON (like an HTML error page), the list should contain None for that specific URL.

Starter Code Boilerplate

import asyncio
import requests
from requests.exceptions import RequestException

def fetch_data_sync(url: str, token: str):
    """
    Standard blocking function to fetch and safely deserialize data.
    """
    # TODO: Create headers with a custom User-Agent and Authorization token
    # TODO: Make a requests.get() call with a timeout
    # TODO: Perform safe deserialization and exception handling
    pass

async def fetch_all_data(urls: list, token: str):
    """
    Asynchronous function to handle multiple blocking requests concurrently.
    """
    # TODO: Create a list of tasks using asyncio.to_thread() to offload the blocking function
    # TODO: Await all tasks concurrently and return the results
    pass

# --- Execution Example ---
# urls_to_fetch = ["https://api.example.com/data1", "https://api.example.com/data2"]
# token = "secret_abc123"
# results = asyncio.run(fetch_all_data(urls_to_fetch, token))
# print(results)

Hints * Bridging Async and Blocking: Standard courses stop at basic asyncio; use the cutting-edge pattern asyncio.to_thread(fetch_data_sync, url, token) inside your async loop; this takes your blocking function and asynchronously runs it in an entirely separate thread yielding control back to the system. * Gathering Tasks: Look into asyncio.gather() to run all your threaded tasks at the exact same time and collect their results into a single list. * Headers: APIs use headers as metadata envelopes; pass a custom User-Agent (e.g., "AnalyticsDashboard/1.0") to bypass bot blockers and pass your token as the "Authorization" header. * Safe Deserialization: Don't really blindly call .json(). Use response.raise_for_status() first to ensure the server actually responded by a success code. * Exception Handling: Catch requests.exceptions.RequestException to handle network and timeout errors and ValueError or requests.exceptions.JSONDecodeError to protect your code from crashing when the server sends back flat HTML instead about JSON.

Test Cases Since we don't have live unreliable API, you can mentally verify or mock the following scenarios:

1. Test Case 1 (All Success):
2. Input: ["url1", "url2"], valid token.
3. Simulated API: Both respond instantly with {"status": "ok", "users": 150}.
4. Expected Output: [{"status": "ok", "users": 150}, {"status": "ok", "users": 150}]
5. Test Case 2 (Timeout & HTML Error):
6. Input: ["url_timeout", "url_html", "url_success"], valid token.
7. Simulated API: URL 1 hangs forever. URL 2 returns an HTML 500 server error page. URL 3 returns {"users": 42}.
8. Expected Output: [None, None, {"users": 42}]
9. Test Case 3 (Missing Headers/Bot Block):
10. Input: ["url_strict"], missing custom User-Agent.
11. Simulated API: Server rejects default Python scripts and returns 403 Forbidden.
12. Expected Output: [None] (Program shouldn't crash just catch the error and return None).