You are building a production-ready async data pipeline that fetches posts and their comments from `jsonplaceholder.typicode.com`. The pipeline must handle concurrency, apply rate limiting, enforce timeouts, and gracefully recover from individual failures.

The following endpoints are available:

- `https://jsonplaceholder.typicode.com/posts/{post_id}` – returns a post object with `title` and `userId` fields;
- `https://jsonplaceholder.typicode.com/comments?postId={post_id}` – returns a list of comments for a given post.



import unittest
import asyncio
import re

# Helper function to dynamically generate test names and assertions
def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)


class TestUserCode(unittest.TestCase):

    def test_fetch_post_is_coroutine(self):
        import user_code
        _dynamic_test(
            self,
            asyncio.iscoroutinefunction(user_code.fetch_post),
            "The `fetch_post` function is defined with `async def`.",
            "Expected `fetch_post` to be defined with `async def`."
        )

    def test_fetch_comment_count_is_coroutine(self):
        import user_code
        _dynamic_test(
            self,
            asyncio.iscoroutinefunction(user_code.fetch_comment_count),
            "The `fetch_comment_count` function is defined with `async def`.",
            "Expected `fetch_comment_count` to be defined with `async def`."
        )

    def test_build_report_is_coroutine(self):
        import user_code
        _dynamic_test(
            self,
            asyncio.iscoroutinefunction(user_code.build_report),
            "The `build_report` function is defined with `async def`.",
            "Expected `build_report` to be defined with `async def`."
        )

    def test_semaphore_used(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            bool(re.search(r"asyncio\.Semaphore\s*\(\s*5\s*\)", source)),
            "`asyncio.Semaphore(5)` is used to limit concurrency.",
            "Expected `asyncio.Semaphore(5)` to be used in `main()`."
        )

    def test_wait_for_used(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            len(re.findall(r"asyncio\.wait_for\s*\(", source)) >= 2,
            "`asyncio.wait_for()` is used in both `fetch_post()` and `fetch_comment_count()`.",
            "Expected `asyncio.wait_for()` to be used at least twice."
        )

    def test_gather_used(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            len(re.findall(r"asyncio\.gather\s*\(", source)) >= 2,
            "`asyncio.gather()` is used in both `build_report()` and `main()`.",
            "Expected `asyncio.gather()` to be used at least twice."
        )

    def test_fetch_post_returns_dict(self):
        import user_code
        import httpx

        async def run():
            semaphore = asyncio.Semaphore(5)
            async with httpx.AsyncClient() as client:
                result = await user_code.fetch_post(client, semaphore, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, dict) and "title" in result and "user_id" in result,
            "`fetch_post()` returns a dict with `title` and `user_id` keys.",
            f"Expected a dict with `title` and `user_id`, but got `{result}`."
        )

    def test_fetch_comment_count_returns_int(self):
        import user_code
        import httpx

        async def run():
            semaphore = asyncio.Semaphore(5)
            async with httpx.AsyncClient() as client:
                result = await user_code.fetch_comment_count(client, semaphore, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, int) and result > 0,
            f"`fetch_comment_count()` returns a positive integer (got `{result}`).",
            f"Expected a positive integer, but got `{result}`."
        )

    def test_build_report_format(self):
        import user_code
        import httpx

        async def run():
            semaphore = asyncio.Semaphore(5)
            async with httpx.AsyncClient() as client:
                result = await user_code.build_report(client, semaphore, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, str) and "[User" in result and "comments" in result,
            "`build_report()` returns a correctly formatted string.",
            f"Expected a string with `[User` and `comments`, but got `{result}`."
        )


if __name__ == "__main__":
    unittest.main()

test_main.py

Master the skills needed to build efficient, concurrent Python applications using asyncio – from coroutines and event loop fundamentals through tasks, synchronization primitives, and real-world debugging techniques.

Gain a solid understanding of asynchronous programming concepts and the core mechanisms behind Python's asyncio library.

Dive into asyncio's task scheduling mechanisms and learn how to run multiple coroutines concurrently.

Get hands-on experience with HTTP clients, synchronization primitives, and queue-based pipelines using asyncio.

Explore advanced asyncio features for controlling task groups, running blocking code, and debugging asynchronous applications.

Challenge: Async Web Scraper with Rate Limiting

Soluzione