You are building a data aggregation script that pulls information about users and their posts from `jsonplaceholder.typicode.com`. Your goal is to fetch all the data concurrently and produce a summary report.

The following endpoints are available:

- `https://jsonplaceholder.typicode.com/users/{user_id}` – returns a user object with `name` and `email` fields;
- `https://jsonplaceholder.typicode.com/posts?userId={user_id}` – returns a list of posts for a given user.



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_user_is_coroutine(self):
        import user_code
        _dynamic_test(
            self,
            asyncio.iscoroutinefunction(user_code.fetch_user),
            "The `fetch_user` function is defined with `async def`.",
            "Expected `fetch_user` to be defined with `async def`."
        )

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

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

    def test_gather_used_in_fetch_user_summary(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 `fetch_user_summary()` and `main()`.",
            "Expected `asyncio.gather()` to be used at least twice – in `fetch_user_summary()` and `main()`."
        )

    def test_fetch_user_returns_dict(self):
        import user_code
        import httpx

        async def run():
            async with httpx.AsyncClient() as client:
                result = await user_code.fetch_user(client, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, dict) and "name" in result and "email" in result,
            "`fetch_user()` returns a dict with `name` and `email` keys.",
            f"Expected `fetch_user()` to return a dict with `name` and `email`, but got `{result}`."
        )

    def test_fetch_post_count_returns_int(self):
        import user_code
        import httpx

        async def run():
            async with httpx.AsyncClient() as client:
                result = await user_code.fetch_post_count(client, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, int) and result > 0,
            f"`fetch_post_count()` returns an integer greater than `0` (got `{result}`).",
            f"Expected `fetch_post_count()` to return a positive integer, but got `{result}`."
        )

    def test_summary_format(self):
        import user_code
        import httpx

        async def run():
            async with httpx.AsyncClient() as client:
                result = await user_code.fetch_user_summary(client, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, str) and "–" in result and "posts" in result,
            f"`fetch_user_summary()` returns a correctly formatted string.",
            f"Expected the summary to contain `–` and `posts`, 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: Fetching Data Concurrently

Рішення