Sample code for the article on asyncio

python-asyncio/README.md

@@ -0,0 +1,3 @@
+# Async I/O in Python: A Complete Walkthrough
+
+This folder provides the code examples for the Real Python tutorial [Async I/O in Python: A Complete Walkthrough](https://realpython.com/async-io-python/).

python-asyncio/as_completed.py

@@ -0,0 +1,21 @@
+import asyncio
+import time
+
+
+async def coro(numbers):
+    await asyncio.sleep(min(numbers))
+    return list(reversed(numbers))
+
+
+async def main():
+    task1 = asyncio.create_task(coro([10, 5, 2]))
+    task2 = asyncio.create_task(coro([3, 2, 1]))
+    print("Start:", time.strftime("%X"))
+    for task in asyncio.as_completed([task1, task2]):
+        result = await task
+        print(f'result: {result} completed at {time.strftime("%X")}')
+    print("End:", time.strftime("%X"))
+    print(f"Both tasks done: {all((task1.done(), task2.done()))}")
+
+
+asyncio.run(main())

python-asyncio/chained.py

@@ -0,0 +1,45 @@
+import asyncio
+import random
+import time
+
+
+async def main():
+    user_ids = [1, 2, 3]
+    start = time.perf_counter()
+    await asyncio.gather(
+        *(get_user_with_posts(user_id) for user_id in user_ids)
+    )
+    end = time.perf_counter()
+    print(f"\n==> Total time: {end - start:.2f} seconds")
+
+
+async def get_user_with_posts(user_id):
+    user = await fetch_user(user_id)
+    await fetch_posts(user)
+
+
+async def fetch_user(user_id):
+    delay = random.uniform(0.5, 2.0)
+    print(f"User coro: fetching user by {user_id=}...")
+    await asyncio.sleep(delay)
+    user = {"id": user_id, "name": f"User{user_id}"}
+    print(f"User coro: fetched user with {user_id=} (done in {delay:.1f}s).")
+    return user
+
+
+async def fetch_posts(user):
+    delay = random.uniform(0.5, 2.0)
+    print(f"Post coro: retrieving posts for {user['name']}...")
+    await asyncio.sleep(delay)
+    posts = [f"Post {i} by {user['name']}" for i in range(1, 3)]
+    print(
+        f"Post coro: got {len(posts)} posts by {user['name']}"
+        f" (done in {delay:.1f}s):"
+    )
+    for post in posts:
+        print(f" - {post}")
+
+
+if __name__ == "__main__":
+    random.seed(444)
+    asyncio.run(main())

python-asyncio/countasync.py

@@ -0,0 +1,21 @@
+import asyncio
+
+
+async def count():
+    print("One")
+    await asyncio.sleep(1)
+    print("Two")
+    await asyncio.sleep(1)
+
+
+async def main():
+    await asyncio.gather(count(), count(), count())
+
+
+if __name__ == "__main__":
+    import time
+
+    start = time.perf_counter()
+    asyncio.run(main())
+    elapsed = time.perf_counter() - start
+    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

python-asyncio/countsync.py

@@ -0,0 +1,20 @@
+import time
+
+
+def count():
+    print("One")
+    time.sleep(1)
+    print("Two")
+    time.sleep(1)
+
+
+def main():
+    for _ in range(3):
+        count()
+
+
+if __name__ == "__main__":
+    start = time.perf_counter()
+    main()
+    elapsed = time.perf_counter() - start
+    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

python-asyncio/except_group.py

@@ -0,0 +1,35 @@
+import asyncio
+
+
+async def coro_a():
+    await asyncio.sleep(1)
+    raise ValueError("Error in coro A")
+
+
+async def coro_b():
+    await asyncio.sleep(2)
+    raise TypeError("Error in coro B")
+
+
+async def coro_c():
+    await asyncio.sleep(0.5)
+    raise IndexError("Error in coro C")
+
+
+async def main():
+    results = await asyncio.gather(
+        coro_a(), coro_b(), coro_c(), return_exceptions=True
+    )
+    exceptions = [e for e in results if isinstance(e, Exception)]
+    if exceptions:
+        raise ExceptionGroup("Errors", exceptions)
+
+
+try:
+    asyncio.run(main())
+except* ValueError as ve_group:
+    print(f"[ValueError handled] {ve_group.exceptions}")
+except* TypeError as te_group:
+    print(f"[TypeError handled] {te_group.exceptions}")
+except* IndexError as ie_group:
+    print(f"[IndexError handled] {ie_group.exceptions}")

python-asyncio/gathers.py

@@ -0,0 +1,21 @@
+import asyncio
+import time
+
+
+async def coro(numbers):
+    await asyncio.sleep(min(numbers))
+    return list(reversed(numbers))
+
+
+async def main():
+    task1 = asyncio.create_task(coro([10, 5, 2]))
+    task2 = asyncio.create_task(coro([3, 2, 1]))
+    print("Start:", time.strftime("%X"))
+    result = await asyncio.gather(task1, task2)
+    print("End:", time.strftime("%X"))
+    print(f"Both tasks done: {all((task1.done(), task2.done()))}")
+    return result
+
+
+result = asyncio.run(main())
+print(f"result: {result}")

python-asyncio/powers.py

@@ -0,0 +1,23 @@
+import asyncio
+
+
+async def powers_of_two(stop=10):
+    exponent = 0
+    while exponent < stop:
+        yield 2**exponent
+        exponent += 1
+        await asyncio.sleep(0.2)  # Simulate some asynchronous work
+
+
+async def main():
+    g = []
+    async for i in powers_of_two(5):
+        g.append(i)
+    print(g)
+
+    f = [j async for j in powers_of_two(5) if not (j // 3 % 5)]
+    print(f)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python-asyncio/queued.py

@@ -0,0 +1,52 @@
+import asyncio
+import random
+import time
+
+
+async def main():
+    queue = asyncio.Queue()
+    user_ids = [1, 2, 3]
+
+    start = time.perf_counter()
+    await asyncio.gather(
+        producer(queue, user_ids),
+        *(consumer(queue) for _ in user_ids),
+    )
+    end = time.perf_counter()
+    print(f"\n==> Total time: {end - start:.2f} seconds")
+
+
+async def producer(queue, user_ids):
+    async def fetch_user(user_id):
+        delay = random.uniform(0.5, 2.0)
+        print(f"Producer: fetching user by {user_id=}...")
+        await asyncio.sleep(delay)
+        user = {"id": user_id, "name": f"User{user_id}"}
+        print(f"Producer: fetched user with {user_id=} (done in {delay:.1f}s)")
+        await queue.put(user)
+
+    await asyncio.gather(*(fetch_user(uid) for uid in user_ids))
+    for _ in range(len(user_ids)):
+        await queue.put(None)  # Sentinels for consumers to terminate
+
+
+async def consumer(queue):
+    while True:
+        user = await queue.get()
+        if user is None:
+            break
+        delay = random.uniform(0.5, 2.0)
+        print(f"Consumer: retrieving posts for {user['name']}...")
+        await asyncio.sleep(delay)
+        posts = [f"Post {i} by {user['name']}" for i in range(1, 3)]
+        print(
+            f"Consumer: got {len(posts)} posts by {user['name']}"
+            f" (done in {delay:.1f}s):"
+        )
+        for post in posts:
+            print(f"  - {post}")
+
+
+if __name__ == "__main__":
+    random.seed(444)
+    asyncio.run(main())

python-asyncio/rand.py

@@ -0,0 +1,34 @@
+import asyncio
+import random
+
+COLORS = (
+    "\033[0m",  # End of color
+    "\033[36m",  # Cyan
+    "\033[91m",  # Red
+    "\033[35m",  # Magenta
+)
+
+
+async def main():
+    return await asyncio.gather(
+        makerandom(1, 9),
+        makerandom(2, 8),
+        makerandom(3, 8),
+    )
+
+
+async def makerandom(delay, threshold=6):
+    color = COLORS[delay]
+    print(f"{color}Initiated makerandom({delay}).")
+    while (number := random.randint(0, 10)) <= threshold:
+        print(f"{color}makerandom({delay}) == {number} too low; retrying.")
+        await asyncio.sleep(delay)
+    print(f"{color}---> Finished: makerandom({delay}) == {number}" + COLORS[0])
+    return number
+
+
+if __name__ == "__main__":
+    random.seed(444)
+    r1, r2, r3 = asyncio.run(main())
+    print()
+    print(f"r1: {r1}, r2: {r2}, r3: {r3}")

python-asyncio/tasks.py

@@ -0,0 +1,17 @@
+import asyncio
+
+
+async def coro(numbers):
+    await asyncio.sleep(min(numbers))
+    return list(reversed(numbers))
+
+
+async def main():
+    task = asyncio.create_task(coro([3, 2, 1]))
+    print(f"{type(task) = }")
+    print(f"{task.done() = }")
+    return await task
+
+
+result = asyncio.run(main())
+print(f"result: {result}")

python-asyncio/websites.py

@@ -0,0 +1,21 @@
+import asyncio
+
+import aiohttp
+
+
+async def check(url):
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url) as response:
+            print(f"{url}: status -> {response.status}")
+
+
+async def main():
+    websites = [
+        "https://realpython.com",
+        "https://pycoders.com",
+        "https://www.python.org",
+    ]
+    await asyncio.gather(*(check(url) for url in websites))
+
+
+asyncio.run(main())