โณ AsyncIO
AsyncIO is a library to write concurrent code using the async/await syntax.
๐ก Quick Tip:
Mastering this concept will significantly boost your Python data science skills!
๐ป Code Example:
import asyncio import time import random # 1. Simple coroutine async def greet(name: str, delay: float) -> str: await asyncio.sleep(delay) # Non-blocking sleep return f"Hello, {name}! (after {delay}s)" result = asyncio.run(greet("santoshtvk", 0.5)) print(result) # 2. gather โ run multiple coroutines concurrently async def fetch_pynfinity_score(user_id: int) -> dict: await asyncio.sleep(random.uniform(0.1, 0.5)) # Simulate I/O return {"user_id": user_id, "score": (user_id * 37) % 100} async def main_gather(): start = time.perf_counter() results = await asyncio.gather( *[fetch_pynfinity_score(i) for i in range(1, 11)] ) elapsed = time.perf_counter() - start # Concurrent: ~0.5s total vs ~3s sequential print(f"\n10 users fetched in {elapsed:.2f}s (concurrently)") for r in results[:5]: print(f" User {r['user_id']:>2}: score={r['score']}") asyncio.run(main_gather()) # 3. Task โ fire and forget async def background_notification(msg: str): await asyncio.sleep(0.3) print(f"[FCM] Sent: {msg}") async def main_tasks(): task1 = asyncio.create_task(background_notification("New pebble! ๐")) task2 = asyncio.create_task(background_notification("Streak at risk! ๐ฅ")) print("Tasks created โ doing other work...") await asyncio.sleep(0.1) await task1; await task2 asyncio.run(main_tasks()) # 4. asyncio.Queue โ producer/consumer pattern async def producer(q: asyncio.Queue): for i in range(5): await q.put(f"job_{i}") await asyncio.sleep(0.1) await q.put(None) # Sentinel to stop consumer async def consumer(q: asyncio.Queue): while True: item = await q.get() if item is None: break print(f" [Consumer] Processing {item}") await asyncio.sleep(0.05) async def main_queue(): q = asyncio.Queue() await asyncio.gather(producer(q), consumer(q)) print("\nQueue demo:") asyncio.run(main_queue())
Keep exploring and happy coding! ๐ป
