Eric Campbell 5 min read Python

Python asyncio Event Loop: Complete Guide

The asyncio event loop is the heart of Python's asynchronous programming model. It's a scheduler that manages the execution of coroutines, callbacks, and I/O operations in a single thread through...

Key Insights

  • The event loop is asyncio’s scheduler that manages coroutine execution through cooperative multitasking, allowing thousands of concurrent operations without thread overhead
  • Modern Python (3.7+) uses asyncio.run() as the primary interface, while direct loop manipulation is reserved for advanced scenarios like thread integration or custom frameworks
  • Blocking operations inside async functions freeze the entire event loop—use run_in_executor() or async libraries to maintain concurrency

Introduction to the Event Loop

The asyncio event loop is the heart of Python’s asynchronous programming model. It’s a scheduler that manages the execution of coroutines, callbacks, and I/O operations in a single thread through cooperative multitasking. Unlike preemptive multithreading where the OS decides when to switch contexts, the event loop relies on coroutines voluntarily yielding control at await points.

Here’s how the event loop operates:

# Event Loop Lifecycle:
# 1. Schedule coroutines and callbacks in a queue
# 2. Execute ready tasks until they hit an await point
# 3. Register I/O operations with the OS (epoll/kqueue/IOCP)
# 4. Wait for I/O completion or timers
# 5. Resume coroutines when their awaited operations complete
# 6. Repeat until no tasks remain

import asyncio

async def main():
    print("Starting")
    await asyncio.sleep(1)  # Yields control to event loop
    print("Finished")

# Modern approach - creates loop, runs main(), closes loop
asyncio.run(main())

When you call asyncio.run(), Python creates an event loop, executes your coroutine, and handles cleanup automatically. This simplicity hides significant complexity underneath.

Getting and Running Event Loops

Python 3.7+ introduced asyncio.run() as the recommended way to run async code. It handles loop creation, execution, and cleanup in one call:

import asyncio

async def fetch_data():
    await asyncio.sleep(0.5)
    return "Data retrieved"

# Modern approach (Python 3.7+)
result = asyncio.run(fetch_data())
print(result)

For legacy code or special cases, you might encounter manual loop management:

# Legacy pattern (pre-3.7 or special requirements)
async def legacy_main():
    await asyncio.sleep(1)
    return "Complete"

loop = asyncio.get_event_loop()
try:
    result = loop.run_until_complete(legacy_main())
    print(result)
finally:
    loop.close()

Creating custom event loop instances is necessary when running asyncio in multiple threads:

import asyncio
import threading

def run_in_thread(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

# Create a new event loop for a specific thread
new_loop = asyncio.new_event_loop()
thread = threading.Thread(target=run_in_thread, args=(new_loop,))
thread.start()

# Schedule work on that loop from another thread
asyncio.run_coroutine_threadsafe(some_coroutine(), new_loop)

Use asyncio.run() for 99% of applications. Only manipulate loops directly when building frameworks, integrating with threaded code, or managing long-lived services.

Scheduling Tasks and Callbacks

The event loop provides multiple methods for scheduling work. Understanding these is crucial for advanced async patterns.

Immediate callback scheduling:

import asyncio

def callback(n):
    print(f"Callback executed with {n}")

async def main():
    loop = asyncio.get_running_loop()
    
    # Schedule callback to run on next loop iteration
    loop.call_soon(callback, 1)
    loop.call_soon(callback, 2)
    
    await asyncio.sleep(0.1)  # Give callbacks time to execute

asyncio.run(main())
# Output:
# Callback executed with 1
# Callback executed with 2

Delayed execution:

import asyncio
import time

def delayed_callback(scheduled_time):
    actual_time = time.time()
    print(f"Scheduled: {scheduled_time:.2f}, Actual: {actual_time:.2f}")

async def schedule_delayed():
    loop = asyncio.get_running_loop()
    
    # Execute after 2 seconds
    loop.call_later(2, delayed_callback, time.time() + 2)
    
    # Execute at specific time
    target_time = loop.time() + 1
    loop.call_at(target_time, lambda: print("Executed at specific time"))
    
    await asyncio.sleep(3)

asyncio.run(schedule_delayed())

Task creation for concurrent execution:

import asyncio

async def worker(name, delay):
    print(f"{name} starting")
    await asyncio.sleep(delay)
    print(f"{name} finished")
    return f"{name} result"

async def main():
    # Create tasks for concurrent execution
    task1 = asyncio.create_task(worker("Task-1", 2))
    task2 = asyncio.create_task(worker("Task-2", 1))
    
    # Tasks run concurrently
    results = await asyncio.gather(task1, task2)
    print(f"Results: {results}")

asyncio.run(main())
# Output shows Task-2 finishes before Task-1

Event Loop Methods and Operations

For long-running services, you need direct control over the event loop lifecycle:

import asyncio
import signal

async def background_task():
    counter = 0
    while True:
        counter += 1
        print(f"Background task iteration {counter}")
        await asyncio.sleep(1)

def shutdown(loop):
    print("Shutting down gracefully...")
    loop.stop()

def run_service():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    # Register signal handlers for graceful shutdown
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, lambda: shutdown(loop))
    
    # Schedule background work
    loop.create_task(background_task())
    
    try:
        print(f"Loop running: {loop.is_running()}")
        loop.run_forever()
    finally:
        print(f"Loop closed: {loop.is_closed()}")
        loop.close()

# Run with Ctrl+C to test graceful shutdown
# run_service()

Checking loop state:

import asyncio

async def check_loop_state():
    loop = asyncio.get_running_loop()
    
    print(f"Is running: {loop.is_running()}")  # True
    print(f"Is closed: {loop.is_closed()}")    # False
    
    # Access loop time (monotonic clock)
    print(f"Loop time: {loop.time()}")

asyncio.run(check_loop_state())

Advanced Patterns: Custom Event Loops and Thread Integration

Running asyncio code from synchronous contexts requires threading:

import asyncio
import threading
from concurrent.futures import Future

class AsyncRunner:
    def __init__(self):
        self.loop = asyncio.new_event_loop()
        self.thread = threading.Thread(target=self._run_loop, daemon=True)
        self.thread.start()
    
    def _run_loop(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()
    
    def run_coroutine(self, coro):
        """Run coroutine from sync code, return Future"""
        return asyncio.run_coroutine_threadsafe(coro, self.loop)
    
    def shutdown(self):
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.thread.join()

# Usage
runner = AsyncRunner()

async def async_operation():
    await asyncio.sleep(1)
    return "Result from async"

future = runner.run_coroutine(async_operation())
result = future.result()  # Blocks until complete
print(result)

runner.shutdown()

Handling blocking operations:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io(duration):
    """Simulates blocking I/O"""
    time.sleep(duration)
    return f"Blocked for {duration}s"

async def main():
    loop = asyncio.get_running_loop()
    
    # Run blocking operation in thread pool
    with ThreadPoolExecutor(max_workers=3) as executor:
        # Execute multiple blocking operations concurrently
        futures = [
            loop.run_in_executor(executor, blocking_io, 2),
            loop.run_in_executor(executor, blocking_io, 1),
            loop.run_in_executor(executor, blocking_io, 3)
        ]
        
        results = await asyncio.gather(*futures)
        print(results)

asyncio.run(main())

Common Pitfalls and Best Practices

Anti-pattern: Blocking the event loop

import asyncio
import time

async def bad_approach():
    # DON'T DO THIS - blocks entire event loop
    time.sleep(2)
    return "Done"

async def good_approach():
    # Use async sleep instead
    await asyncio.sleep(2)
    return "Done"

# With bad_approach, nothing else can run during sleep

Anti-pattern: Nested asyncio.run()

import asyncio

async def outer():
    # NEVER DO THIS - raises RuntimeError
    # asyncio.run(inner())
    
    # Instead, just await the coroutine
    await inner()

async def inner():
    return "Success"

asyncio.run(outer())

Debugging with debug mode:

import asyncio
import logging

# Enable debug mode for detailed warnings
logging.basicConfig(level=logging.DEBUG)

async def slow_callback():
    await asyncio.sleep(0.15)  # Will trigger slow callback warning

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

Practical Example: Building a Concurrent Web Scraper

Here’s a complete web scraper demonstrating event loop concepts:

import asyncio
import aiohttp
from typing import List, Dict
import time

class AsyncScraper:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []
    
    async def fetch_url(self, session: aiohttp.ClientSession, url: str) -> Dict:
        async with self.semaphore:  # Limit concurrent requests
            try:
                start = time.time()
                async with session.get(url, timeout=10) as response:
                    content = await response.text()
                    duration = time.time() - start
                    
                    return {
                        'url': url,
                        'status': response.status,
                        'size': len(content),
                        'duration': duration
                    }
            except Exception as e:
                return {'url': url, 'error': str(e)}
    
    async def scrape_urls(self, urls: List[str]) -> List[Dict]:
        async with aiohttp.ClientSession() as session:
            tasks = [
                asyncio.create_task(self.fetch_url(session, url))
                for url in urls
            ]
            
            # Gather with progress reporting
            for i, task in enumerate(asyncio.as_completed(tasks)):
                result = await task
                self.results.append(result)
                print(f"Completed {i+1}/{len(tasks)}: {result.get('url')}")
            
            return self.results

async def main():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/status/404',
        'https://httpbin.org/delay/3',
    ]
    
    scraper = AsyncScraper(max_concurrent=3)
    start = time.time()
    
    results = await scraper.scrape_urls(urls)
    
    duration = time.time() - start
    print(f"\nScraped {len(results)} URLs in {duration:.2f}s")
    
    # Show statistics
    successful = [r for r in results if 'error' not in r]
    print(f"Success rate: {len(successful)}/{len(results)}")

if __name__ == "__main__":
    asyncio.run(main())

This scraper demonstrates task creation, semaphore-based concurrency limiting, proper error handling, and real-world async patterns. The event loop manages all concurrent requests efficiently in a single thread, achieving performance that would require complex thread pool management in synchronous code.

Understanding the event loop transforms asyncio from a mysterious abstraction into a powerful tool for building high-performance I/O-bound applications.

Liked this? There's more.

Every week: one practical technique, explained simply, with code you can use immediately.