Sync asyncio-walkthrough with the maintenance update of Python's asyncio: A Hands-On Walkthrough (#868)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: realpython-bot

asyncio-walkthrough/areq.py

@@ -1,51 +1,45 @@
-#!/usr/env/bin python3
-# chained.py
-
 import asyncio
 import random
 import time
 
 
-async def randint(a: int, b: int) -> int:
-    return random.randint(a, b)
-
+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 part1(n: int) -> str:
-    i = await randint(0, 10)
-    print(f"part1({n}) sleeping for {i} seconds.")
-    await asyncio.sleep(i)
-    result = f"result{n}-1"
-    print(f"Returning part1({n}) == {result}.")
-    return result
 
+async def get_user_with_posts(user_id):
+    user = await fetch_user(user_id)
+    await fetch_posts(user)
 
-async def part2(n: int, arg: str) -> str:
-    i = await randint(0, 10)
-    print(f"part2{n, arg} sleeping for {i} seconds.")
-    await asyncio.sleep(i)
-    result = f"result{n}-2 derived from {arg}"
-    print(f"Returning part2{n, arg} == {result}.")
-    return result
 
+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 chain(n: int) -> None:
-    start = time.perf_counter()
-    p1 = await part1(n)
-    p2 = await part2(n, p1)
-    end = time.perf_counter() - start
-    print(f"-->Chained result{n} => {p2} (took {end:0.2f} seconds).")
 
-
-async def main(*args):
-    await asyncio.gather(*(chain(n) for n in args))
+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__":
-    import sys
-
     random.seed(444)
-    args = [1, 2, 3] if len(sys.argv) == 1 else map(int, sys.argv[1:])
-    start = time.perf_counter()
-    asyncio.run(main(*args))
-    end = time.perf_counter() - start
-    print(f"Program finished in {end:0.2f} seconds.")
+    asyncio.run(main())

asyncio-walkthrough/asyncq.py

@@ -1,13 +1,11 @@
-#!/usr/bin/env python3
-# countasync.py
-
 import asyncio
 
 
 async def count():
     print("One")
     await asyncio.sleep(1)
     print("Two")
+    await asyncio.sleep(1)
 
 
 async def main():
@@ -17,7 +15,7 @@ async def main():
 if __name__ == "__main__":
     import time
 
-    s = time.perf_counter()
+    start = time.perf_counter()
     asyncio.run(main())
-    elapsed = time.perf_counter() - s
+    elapsed = time.perf_counter() - start
     print(f"{__file__} executed in {elapsed:0.2f} seconds.")

asyncio-walkthrough/chained.py

@@ -1,13 +1,11 @@
-#!/usr/bin/env python3
-# countsync.py
-
 import time
 
 
 def count():
     print("One")
     time.sleep(1)
     print("Two")
+    time.sleep(1)
 
 
 def main():
@@ -16,7 +14,7 @@ def main():
 
 
 if __name__ == "__main__":
-    s = time.perf_counter()
+    start = time.perf_counter()
     main()
-    elapsed = time.perf_counter() - s
+    elapsed = time.perf_counter() - start
     print(f"{__file__} executed in {elapsed:0.2f} seconds.")