When you want to remind yourself or someone else about different tasks at different times, you do not need to wait for one reminder to finish before scheduling the next. Using Python's `asyncio`, you can schedule multiple reminders to run independently, each waiting for its own specific interval before displaying a message. Each reminder is represented as an asynchronous function that sleeps for a certain amount of time and then prints a message. By launching each reminder with `asyncio.create_task`, you allow all reminders to count down simultaneously, so that each message appears at its scheduled time, regardless of when the others are set to trigger. This approach is especially useful for building notification systems, alerts, or any application where you need to handle multiple timed events without blocking your program.


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

class TestTask(unittest.TestCase):
    def test_reminder_waits_and_prints(self):
        import user_code
        importlib.reload(user_code)
        messages = [
            ("Take a break!", 1.0),
            ("Check your email.", 1.5),
            ("Meeting in 5 minutes.", 0.5)
        ]
        for msg, delay in messages:
            f = io.StringIO()
            start = time.monotonic()
            async def run_one():
                with redirect_stdout(f):
                    await user_code.reminder(msg, delay)
            asyncio.run(run_one())
            end = time.monotonic()
            output = normalize_text(f.getvalue())
            _dynamic_test(
                self,
                msg.lower() in output,
                f"Reminder prints: {msg}",
                f"Expected reminder message '{msg}' not found in output: {output}",
            )
            _dynamic_test(
                self,
                end - start >= delay,
                f"Reminder waits at least {delay} seconds before printing.",
                f"Reminder for '{msg}' did not wait {delay}s. Waited: {end-start}s",
            )

    def test_reminders_run_concurrently(self):
        import user_code
        importlib.reload(user_code)
        f = io.StringIO()
        start = time.monotonic()
        with redirect_stdout(f):
            asyncio.run(user_code.main())
        end = time.monotonic()
        output = normalize_text(f.getvalue())
        expected_msgs = [
            "Take a break!",
            "Check your email.",
            "Meeting in 5 minutes."
        ]
        for msg in expected_msgs:
            _dynamic_test(
                self,
                normalize_text(msg) in output,
                f"Output includes reminder: {msg}",
                f"Expected reminder '{msg}' not found in output: {output}",
            )
        duration = end - start
        _dynamic_test(
            self,
            duration >= 3.8 and duration < 6,
            "Reminders run concurrently, total time matches max delay.",
            f"Reminders did not run concurrently. Duration: {duration}s (expected ~4s, not ~7s)",
        )

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 Reminder System

Lösung

Challenge: Async Reminder System

Lösung