In many real-world applications, you need to fetch data from several sources at once. Imagine you are building a weather dashboard that shows the current temperature for multiple cities. Instead of waiting for each city’s data to arrive one after another, you can use Python’s `asyncio` to fetch all the data concurrently, making your program much faster and more responsive. By simulating each weather data fetch as an asynchronous function that waits a random amount of time, you can see how `asyncio.gather` lets you run all these tasks together and collect their results as soon as they’re ready.


import unittest
import user_code
import ast
import re   
import unittest
import importlib
import io
from contextlib import redirect_stdout
import re

class TestTask(unittest.TestCase):
    def test_fetch_weather_format(self):
        import user_code
        import asyncio
        import re
        
        async def test():
            result = await user_code.fetch_weather("Paris")
            pattern = r"^Weather for Paris: (\d{2})°C$"
            match = re.match(pattern, result)
            _dynamic_test(
                self,
                match is not None and 15 <= int(match.group(1)) <= 30,
                "fetch_weather returns correct format and temperature range",
                f"fetch_weather did not return expected format: '{result}'",
            )
        asyncio.run(test())

   

    def test_report_weather_gather_concurrency(self):
        # This test checks that asyncio.gather is used in report_weather
        with open("user_code.py", "r") as f:
            code = f.read()
        _dynamic_test(
            self,
            "asyncio.gather" in code,
            "asyncio.gather is used in report_weather",
            "asyncio.gather not found in report_weather, required for concurrency",
        )

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)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    for i, node in enumerate(tree.body):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == var_name:
                    start_line = node.lineno - 1
                    line = lines[start_line]
                    indent = ' ' * (len(line) - len(line.lstrip()))
                    lines[start_line] = f"{indent}{var_name} = {value}"
                    next_line = len(lines)
                    for next_node in tree.body[i+1:]:
                        if hasattr(next_node, 'lineno'):
                            next_line = next_node.lineno - 1
                            break
                    if next_line > start_line + 1:
                        lines[start_line+1:next_line] = []
                    
                    return '\\n'.join(lines)
    return code

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


test_main.py

A beginner-friendly course introducing Python's asyncio library, focusing on asynchronous programming concepts, practical usage, and hands-on challenges with real-world scenarios.

Introduction to asynchronous programming and the basics of Python's asyncio library.

Delve into creating, running, and managing multiple asyncio tasks concurrently.

Explore more advanced asyncio patterns such as task cancellation and error handling.

Challenge: Async Weather Reporter

Lösung

Challenge: Async Weather Reporter

Lösung