🚀 Asyncio: Event Loop and Tasks

🎯 Introduction

Ever felt like your Python program is stuck waiting for one thing to finish before moving to the next? 🐌 Imagine you’re a chef 👨‍🍳 in a restaurant - would you wait for the soup to boil before starting to chop vegetables? Of course not! You’d multitask! That’s exactly what asyncio helps us do in Python! 🎉

In this tutorial, we’ll explore the magical world of asyncio - Python’s built-in library for writing concurrent code. You’ll learn how to make your programs do multiple things at once, like a master chef juggling multiple dishes! 🍳✨

📚 Understanding Asyncio: The Event Loop and Tasks

Think of asyncio like a restaurant kitchen 🍽️:

• Event Loop = The head chef coordinating everything 👨‍🍳
• Tasks = Individual cooking activities (boiling pasta, grilling steak, etc.) 🍝🥩
• Await = Waiting for something to finish (like the oven timer) ⏰

The event loop is the heart of asyncio - it’s like a super-efficient manager that keeps track of all your tasks and switches between them whenever one is waiting for something! 🔄

The Magic of Concurrency 🎩

# 🐌 Without asyncio (synchronous)
import time

def make_coffee():
    print("☕ Starting coffee...")
    time.sleep(3)  # Waiting 3 seconds
    print("☕ Coffee ready!")

def make_toast():
    print("🍞 Starting toast...")
    time.sleep(2)  # Waiting 2 seconds
    print("🍞 Toast ready!")

# This takes 5 seconds total! 😴
make_coffee()
make_toast()

# 🚀 With asyncio (asynchronous)
import asyncio

async def make_coffee():
    print("☕ Starting coffee...")
    await asyncio.sleep(3)  # Non-blocking wait!
    print("☕ Coffee ready!")

async def make_toast():
    print("🍞 Starting toast...")
    await asyncio.sleep(2)  # Non-blocking wait!
    print("🍞 Toast ready!")

# This takes only 3 seconds! 🎉
async def make_breakfast():
    await asyncio.gather(make_coffee(), make_toast())

asyncio.run(make_breakfast())

🔧 Basic Syntax and Usage

Let’s break down the essential asyncio components! 🧩

1. Creating Async Functions (Coroutines) 🌟

# ✅ Async function (coroutine)
async def greet_user(name):
    print(f"👋 Hello, {name}!")
    await asyncio.sleep(1)  # Simulate some async work
    return f"Nice to meet you, {name}! 🎉"

# ❌ Common mistake - forgetting 'async'
def greet_user_wrong(name):
    print(f"👋 Hello, {name}!")
    await asyncio.sleep(1)  # SyntaxError! 💥
    return f"Nice to meet you, {name}!"

2. Running the Event Loop 🏃‍♂️

import asyncio

async def main():
    # Your async code goes here! 🎯
    result = await greet_user("Python Learner")
    print(result)

# 🚀 Modern way (Python 3.7+)
asyncio.run(main())

# 📚 Old way (still works)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

3. Creating and Managing Tasks 📋

import asyncio

async def fetch_data(url, delay):
    print(f"🌐 Fetching {url}...")
    await asyncio.sleep(delay)  # Simulate network delay
    return f"Data from {url} 📦"

async def main():
    # Create tasks - they start running immediately! 🏃‍♂️
    task1 = asyncio.create_task(fetch_data("api.com", 2))
    task2 = asyncio.create_task(fetch_data("database.com", 3))
    
    # Wait for both tasks to complete
    result1 = await task1
    result2 = await task2
    
    print(f"Got: {result1}")
    print(f"Got: {result2}")

asyncio.run(main())

💡 Practical Examples

Example 1: Web Scraper with Rate Limiting 🕷️

import asyncio
import random

class AsyncWebScraper:
    def __init__(self, max_concurrent=3):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []
    
    async def fetch_page(self, url):
        async with self.semaphore:  # Limit concurrent requests 🚦
            print(f"🌐 Fetching {url}...")
            
            # Simulate network delay
            delay = random.uniform(0.5, 2.0)
            await asyncio.sleep(delay)
            
            # Simulate page content
            content = f"Content from {url} (took {delay:.2f}s)"
            print(f"✅ Done: {url}")
            return content
    
    async def scrape_all(self, urls):
        tasks = [self.fetch_page(url) for url in urls]
        self.results = await asyncio.gather(*tasks)
        return self.results

# Let's scrape some "websites"! 🎉
async def main():
    urls = [
        "https://example1.com",
        "https://example2.com",
        "https://example3.com",
        "https://example4.com",
        "https://example5.com"
    ]
    
    scraper = AsyncWebScraper(max_concurrent=2)
    results = await scraper.scrape_all(urls)
    
    print("\n📊 Results:")
    for result in results:
        print(f"  • {result}")

asyncio.run(main())

Example 2: Real-Time Chat Server 💬

import asyncio
from datetime import datetime

class ChatRoom:
    def __init__(self):
        self.messages = []
        self.users = {}
    
    async def add_user(self, user_id, writer):
        self.users[user_id] = writer
        await self.broadcast(f"👋 {user_id} joined the chat!")
    
    async def remove_user(self, user_id):
        if user_id in self.users:
            del self.users[user_id]
            await self.broadcast(f"👋 {user_id} left the chat!")
    
    async def broadcast(self, message):
        timestamp = datetime.now().strftime("%H:%M:%S")
        formatted_msg = f"[{timestamp}] {message}"
        self.messages.append(formatted_msg)
        
        # Send to all connected users 📡
        tasks = []
        for user_id, writer in self.users.items():
            tasks.append(self.send_to_user(writer, formatted_msg))
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def send_to_user(self, writer, message):
        try:
            writer.write(f"{message}\n".encode())
            await writer.drain()
        except:
            pass  # Handle disconnected users gracefully

# Simulate a chat session! 🎮
async def simulate_chat():
    chat = ChatRoom()
    
    # Simulate users joining
    print("🏛️ Chat Room Opened!")
    
    # Mock writers (in real app, these would be network connections)
    class MockWriter:
        def write(self, data):
            print(f"📨 Sending: {data.decode().strip()}")
        async def drain(self):
            pass
    
    # Simulate chat activity
    await chat.add_user("Alice", MockWriter())
    await asyncio.sleep(0.5)
    
    await chat.add_user("Bob", MockWriter())
    await asyncio.sleep(0.5)
    
    await chat.broadcast("Alice: Hey everyone! 👋")
    await asyncio.sleep(0.5)
    
    await chat.broadcast("Bob: Hi Alice! How are you? 😊")
    await asyncio.sleep(0.5)
    
    await chat.remove_user("Alice")

asyncio.run(simulate_chat())

Example 3: Task Queue with Worker Pool 🏭

import asyncio
import random

class TaskQueue:
    def __init__(self, num_workers=3):
        self.queue = asyncio.Queue()
        self.num_workers = num_workers
        self.results = []
    
    async def worker(self, worker_id):
        """Worker coroutine that processes tasks 🔧"""
        while True:
            # Get a task from the queue
            task = await self.queue.get()
            
            if task is None:  # Poison pill to stop worker
                self.queue.task_done()
                break
            
            print(f"🤖 Worker {worker_id}: Processing {task['name']}...")
            
            # Simulate work
            await asyncio.sleep(task['duration'])
            result = f"Task {task['name']} completed by Worker {worker_id}"
            self.results.append(result)
            
            print(f"✅ Worker {worker_id}: Finished {task['name']}")
            self.queue.task_done()
    
    async def add_task(self, name, duration):
        """Add a task to the queue 📥"""
        await self.queue.put({'name': name, 'duration': duration})
    
    async def process_all(self, tasks):
        """Process all tasks with worker pool 🏊‍♂️"""
        # Start workers
        workers = [
            asyncio.create_task(self.worker(i))
            for i in range(self.num_workers)
        ]
        
        # Add all tasks to queue
        for task in tasks:
            await self.add_task(task['name'], task['duration'])
        
        # Wait for all tasks to be processed
        await self.queue.join()
        
        # Stop workers
        for _ in range(self.num_workers):
            await self.queue.put(None)
        
        # Wait for workers to finish
        await asyncio.gather(*workers)
        
        return self.results

# Let's process some tasks! 🚀
async def main():
    tasks = [
        {'name': 'Download File A', 'duration': 2},
        {'name': 'Process Image B', 'duration': 1.5},
        {'name': 'Send Email C', 'duration': 0.5},
        {'name': 'Generate Report D', 'duration': 3},
        {'name': 'Backup Database E', 'duration': 2.5},
    ]
    
    print("🏭 Starting task processing...")
    queue = TaskQueue(num_workers=2)
    results = await queue.process_all(tasks)
    
    print("\n📊 All tasks completed!")
    for result in results:
        print(f"  • {result}")

asyncio.run(main())

🚀 Advanced Concepts

1. Task Cancellation and Timeouts ⏱️

import asyncio

async def long_running_task():
    try:
        print("🏃‍♂️ Starting long task...")
        await asyncio.sleep(10)
        print("✅ Task completed!")
        return "Success"
    except asyncio.CancelledError:
        print("❌ Task was cancelled!")
        # Cleanup code here
        raise  # Re-raise to properly handle cancellation

async def main():
    # Create a task
    task = asyncio.create_task(long_running_task())
    
    # Cancel after 2 seconds
    await asyncio.sleep(2)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("🎯 Successfully cancelled the task")
    
    # Using timeout
    print("\n⏰ Trying with timeout...")
    try:
        async with asyncio.timeout(2):  # Python 3.11+
            await long_running_task()
    except asyncio.TimeoutError:
        print("⏱️ Task timed out!")

# For Python < 3.11, use wait_for
async def main_legacy():
    try:
        await asyncio.wait_for(long_running_task(), timeout=2)
    except asyncio.TimeoutError:
        print("⏱️ Task timed out!")

asyncio.run(main())

2. Task Groups and Exception Handling 🛡️

import asyncio

async def risky_task(task_id, should_fail=False):
    print(f"🎲 Task {task_id} starting...")
    await asyncio.sleep(1)
    
    if should_fail:
        raise ValueError(f"Task {task_id} failed! 💥")
    
    return f"Task {task_id} succeeded! ✅"

async def main():
    # Using TaskGroup (Python 3.11+)
    try:
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(risky_task(1))
            task2 = tg.create_task(risky_task(2, should_fail=True))
            task3 = tg.create_task(risky_task(3))
    except* ValueError as e:
        print(f"🚨 Some tasks failed: {e.exceptions}")
    
    # For Python < 3.11, use gather with return_exceptions
    results = await asyncio.gather(
        risky_task(1),
        risky_task(2, should_fail=True),
        risky_task(3),
        return_exceptions=True
    )
    
    for i, result in enumerate(results, 1):
        if isinstance(result, Exception):
            print(f"❌ Task {i} error: {result}")
        else:
            print(f"✅ Task {i} result: {result}")

# Run with appropriate version
if hasattr(asyncio, 'TaskGroup'):
    asyncio.run(main())
else:
    # Fallback for older Python versions
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

3. Event Loop Customization 🎛️

import asyncio
import time

class TimedEventLoop(asyncio.BaseEventLoop):
    """Custom event loop that tracks execution time ⏱️"""
    
    def __init__(self):
        super().__init__()
        self.task_times = {}
    
    def create_task(self, coro, *, name=None):
        task = super().create_task(coro, name=name)
        self.task_times[task] = time.time()
        
        # Add callback to measure completion time
        task.add_done_callback(self._task_done)
        return task
    
    def _task_done(self, task):
        if task in self.task_times:
            duration = time.time() - self.task_times[task]
            print(f"⏱️ Task {task.get_name()} took {duration:.2f}s")

# More practical: Event loop with debug mode
async def debug_example():
    # Enable debug mode
    loop = asyncio.get_running_loop()
    loop.set_debug(True)
    
    async def slow_operation():
        # This will trigger a warning in debug mode!
        time.sleep(0.1)  # ❌ Blocking call!
        
    async def fast_operation():
        # This is correct ✅
        await asyncio.sleep(0.1)
    
    # Run both
    await fast_operation()
    # await slow_operation()  # Uncomment to see debug warning

# Run with debug mode
asyncio.run(debug_example(), debug=True)

⚠️ Common Pitfalls and Solutions

1. Forgetting to await 😴

# ❌ Wrong - coroutine not awaited
async def get_data():
    return "Important data! 📊"

async def main_wrong():
    data = get_data()  # This returns a coroutine object!
    print(data)  # Prints: <coroutine object get_data at ...>

# ✅ Correct - properly awaited
async def main_correct():
    data = await get_data()  # Now we get the actual data!
    print(data)  # Prints: Important data! 📊

2. Blocking the Event Loop 🚫

import time
import asyncio

# ❌ Wrong - blocks the event loop
async def bad_sleep():
    print("😴 Sleeping (blocking)...")
    time.sleep(2)  # This blocks everything!
    print("😴 Awake!")

# ✅ Correct - non-blocking sleep
async def good_sleep():
    print("😴 Sleeping (async)...")
    await asyncio.sleep(2)  # Other tasks can run!
    print("😴 Awake!")

# ✅ For CPU-bound tasks, use executor
async def cpu_intensive_task():
    loop = asyncio.get_running_loop()
    
    # Run in a separate thread
    result = await loop.run_in_executor(
        None,  # Use default executor
        lambda: sum(i**2 for i in range(1000000))  # CPU-bound work
    )
    
    return result

3. Creating Tasks Incorrectly 🔧

# ❌ Wrong - tasks not running concurrently
async def fetch_all_wrong(urls):
    results = []
    for url in urls:
        result = await fetch_data(url)  # Sequential! 🐌
        results.append(result)
    return results

# ✅ Correct - tasks run concurrently
async def fetch_all_correct(urls):
    tasks = [fetch_data(url) for url in urls]
    return await asyncio.gather(*tasks)  # Concurrent! 🚀

# ✅ Even better - with create_task
async def fetch_all_better(urls):
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    return await asyncio.gather(*tasks)

🛠️ Best Practices

1. Use Context Managers 🎯

import asyncio
import aiofiles  # pip install aiofiles

# ✅ Good - automatic cleanup
async def read_file_async(filename):
    async with aiofiles.open(filename, 'r') as file:
        content = await file.read()
        return content

# ✅ For network connections
import aiohttp  # pip install aiohttp

async def fetch_json(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

2. Structure Your Async Code 🏗️

class AsyncService:
    """Well-structured async service 🎯"""
    
    def __init__(self):
        self.is_running = False
        self.tasks = set()
    
    async def start(self):
        """Start the service 🚀"""
        self.is_running = True
        print("🟢 Service started!")
    
    async def stop(self):
        """Gracefully stop the service 🛑"""
        self.is_running = False
        
        # Cancel all pending tasks
        for task in self.tasks:
            task.cancel()
        
        # Wait for all tasks to complete
        await asyncio.gather(*self.tasks, return_exceptions=True)
        self.tasks.clear()
        
        print("🔴 Service stopped!")
    
    async def add_background_task(self, coro):
        """Add a managed background task 📋"""
        task = asyncio.create_task(coro)
        self.tasks.add(task)
        task.add_done_callback(self.tasks.discard)
        return task
    
    async def __aenter__(self):
        await self.start()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.stop()

# Usage
async def main():
    async with AsyncService() as service:
        # Service is running! 🏃‍♂️
        await service.add_background_task(some_background_work())
        # Do other stuff...
    # Service automatically stopped! 🛑

3. Handle Exceptions Properly 🛡️

import asyncio
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def resilient_task(task_id):
    """Task with proper error handling 🛡️"""
    try:
        logger.info(f"🚀 Task {task_id} starting...")
        
        # Simulate work that might fail
        if task_id == 2:
            raise ValueError("Simulated error! 💥")
        
        await asyncio.sleep(1)
        return f"Task {task_id} completed! ✅"
        
    except asyncio.CancelledError:
        logger.warning(f"⚠️ Task {task_id} cancelled!")
        # Cleanup if needed
        raise  # Always re-raise CancelledError
        
    except Exception as e:
        logger.error(f"❌ Task {task_id} failed: {e}")
        # Could retry, log to monitoring, etc.
        return f"Task {task_id} failed: {e}"

async def main():
    # Create tasks with proper exception handling
    tasks = [resilient_task(i) for i in range(5)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Process results
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.error(f"Task {i} exception: {result}")
        else:
            logger.info(f"Task {i} result: {result}")

asyncio.run(main())

🧪 Hands-On Exercise

Ready to put your asyncio skills to the test? Let’s build a mini web crawler! 🕷️

Challenge: Create an async web crawler that:

1. Fetches multiple “web pages” concurrently
2. Extracts “links” from each page
3. Follows links up to a certain depth
4. Respects rate limiting (max 3 concurrent requests)
5. Handles errors gracefully

import asyncio
import random
from typing import Set, List

# Your task: Complete this web crawler! 🎯

class AsyncWebCrawler:
    def __init__(self, max_concurrent=3, max_depth=2):
        # TODO: Initialize the crawler
        pass
    
    async def fetch_page(self, url: str) -> dict:
        """Simulate fetching a web page"""
        # TODO: Implement rate-limited page fetching
        # Should return {'url': url, 'links': [...], 'content': '...'}
        pass
    
    async def crawl(self, start_url: str) -> List[dict]:
        """Crawl starting from the given URL"""
        # TODO: Implement the crawling logic
        # Should handle multiple depths and avoid duplicate URLs
        pass

# Test your crawler!
async def main():
    crawler = AsyncWebCrawler(max_concurrent=3, max_depth=2)
    results = await crawler.crawl("https://example.com")
    
    print(f"🕷️ Crawled {len(results)} pages!")
    for page in results:
        print(f"  • {page['url']} ({len(page['links'])} links)")

# asyncio.run(main())

import asyncio
import random
from typing import Set, List, Dict

class AsyncWebCrawler:
    def __init__(self, max_concurrent=3, max_depth=2):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.max_depth = max_depth
        self.visited_urls: Set[str] = set()
        self.results: List[Dict] = []
    
    async def fetch_page(self, url: str) -> dict:
        """Simulate fetching a web page 🌐"""
        async with self.semaphore:  # Rate limiting! 🚦
            print(f"🔍 Fetching {url}...")
            
            # Simulate network delay
            await asyncio.sleep(random.uniform(0.5, 1.5))
            
            # Simulate page content with random links
            num_links = random.randint(2, 5)
            links = [
                f"{url}/page{i}" 
                for i in range(num_links)
            ]
            
            # Simulate occasional errors
            if random.random() < 0.1:  # 10% chance of error
                raise Exception(f"Failed to fetch {url}! 💥")
            
            page_data = {
                'url': url,
                'links': links,
                'content': f'Content of {url} 📄'
            }
            
            print(f"✅ Fetched {url}")
            return page_data
    
    async def crawl_url(self, url: str, depth: int):
        """Crawl a single URL at given depth 🕷️"""
        if depth > self.max_depth or url in self.visited_urls:
            return
        
        self.visited_urls.add(url)
        
        try:
            page = await self.fetch_page(url)
            self.results.append(page)
            
            # Crawl links if not at max depth
            if depth < self.max_depth:
                tasks = [
                    self.crawl_url(link, depth + 1)
                    for link in page['links']
                    if link not in self.visited_urls
                ]
                await asyncio.gather(*tasks, return_exceptions=True)
                
        except Exception as e:
            print(f"❌ Error crawling {url}: {e}")
    
    async def crawl(self, start_url: str) -> List[dict]:
        """Start crawling from the given URL 🚀"""
        print(f"🕷️ Starting crawl from {start_url}")
        print(f"📊 Max depth: {self.max_depth}, Max concurrent: {self.semaphore._value}")
        
        await self.crawl_url(start_url, depth=0)
        
        print(f"\n✨ Crawl complete! Visited {len(self.visited_urls)} URLs")
        return self.results

# Test the crawler! 🎮
async def main():
    crawler = AsyncWebCrawler(max_concurrent=3, max_depth=2)
    results = await crawler.crawl("https://example.com")
    
    print(f"\n🕷️ Crawled {len(results)} pages successfully!")
    for page in results:
        print(f"  • {page['url']} ({len(page['links'])} links found)")

asyncio.run(main())

🎓 Key Takeaways

Congratulations! You’ve mastered asyncio fundamentals! 🎉 Here’s what you’ve learned:

1. Event Loop 🔄 - The conductor orchestrating your async symphony
2. Coroutines & Tasks 🎯 - Your async building blocks for concurrent execution
3. Concurrency Patterns 🚀 - Running multiple operations without blocking
4. Error Handling 🛡️ - Gracefully managing failures in async code
5. Best Practices 🏆 - Writing clean, efficient, and maintainable async code

Remember:

• Use async/await for I/O-bound operations 🌐
• Don’t block the event loop with synchronous code 🚫
• Always handle exceptions properly 🛡️
• Use context managers for resource cleanup 🧹

🤝 Next Steps

You’re now ready to build amazing concurrent applications! Here’s what to explore next:

1. 🔧 AsyncIO Streams - Network programming with asyncio
2. 🔄 Synchronization Primitives - Locks, events, and conditions
3. 🌐 Async Web Frameworks - FastAPI, aiohttp, Quart
4. 📊 Async Database Operations - asyncpg, motor, aioredis
5. 🎮 Real-time Applications - WebSockets, chat servers, game servers

Keep practicing, and remember: concurrency is like juggling 🤹‍♂️ - it gets easier with practice! You’ve got this! 💪

Happy async coding! 🚀✨