Concurrency in Python

Nazneen Mulani — Tue, 06 May 2025 06:39:34 +0000

Fundamentals of Concurrency

Concurrency: Managing multiple tasks simultaneously to improve responsiveness.

Threads and asyncio operate on a single processor, switching tasks.
Multiprocessing achieves true parallelism by using multiple CPU cores.

Parallelism: Running tasks literally at the same time on different processors.

Types of Tasks:

I/O-Bound: Limited by input/output operations (e.g., network, file system).
CPU-Bound: Limited by the processor's computational power.

Concurrency Models

Threading:

Use for I/O-bound tasks.
Managed by the OS (preemptive multitasking).
Python module: threading, concurrent.futures.ThreadPoolExecutor.

Asyncio:

Best for I/O-bound tasks.
Uses cooperative multitasking (tasks decide when to yield control).
Requires non-blocking libraries (e.g., aiohttp for HTTP requests).
Python module: asyncio.

Multiprocessing:

Use for CPU-bound tasks.
Achieves parallelism by using multiple processes.
Python module: multiprocessing, concurrent.futures.ProcessPoolExecutor.

Advantages and Limitations

Threading:

✅ Simplifies concurrency for I/O-bound tasks.

❌ Limited by Python’s Global Interpreter Lock (GIL) for CPU-bound tasks.

Asyncio:

✅ Lightweight; excellent for scaling I/O-bound tasks.

❌ Requires non-blocking libraries; prone to errors if blocking calls are used.

Multiprocessing:

✅ Best for CPU-bound tasks; achieves true parallelism.

❌ Higher overhead for creating processes.

asyncio — Asynchronous I/O

As per official document of Python (version 3.13.3 when writing this blog) below are the key concepts to remember when working with asyncio in Python -

Summary:
asyncio is a library to write concurrent code using the async/await syntax.
asyncio is used as a foundation for multiple Python asynchronous frameworks that provide high-performance network and web-servers, database connection libraries, distributed task queues, etc.
asyncio is often a perfect fit for IO-bound and high-level structured network code.

import asyncio

async def main():
    print('Guten ...')
    await asyncio.sleep(1)
    print('... morgen!')

asyncio.run(main())

asyncio provides a set of high-level and low-level APIs to make use of it as a Python developer.

High level APIs include Runners, Coroutines and Tasks, Streams, Synchronization Primitives, Subprocesses, Queues, Exceptions.
While Low level APIs include Event Loop, Futures, Transports and, Protocols, Policies, Platform Support, Extending .

Basic terms to remember -

AsyncIO Runner
Event Loop
Coroutine
Task Queue
Future Results

AsyncIO Runner
This is the outer container that manages the entire async execution
Creates and manages the event loop
Handles setup and cleanup of the async environment

Event Loop
The engine that drives asynchronous execution
Continuously runs in cycles, checking for tasks to execute
Manages the scheduling of coroutines
Tracks when I/O operations complete

Coroutine
A function marked with async def
Can be paused and resumed using await

Task Queue
Where the event loop stores tasks that are ready to run
Manages the order of execution

Future Results
Stores the eventual results of async operations
Used to track completion status

Runner API

asyncio.run(coro, *, debug=None, loop_factory=None)

Execute the coroutine coro and return the result.

Example,

async def main():
    await asyncio.sleep(1)
    print('hey! how are are you?')

asyncio.run(main())

Runner Context Manager API
class asyncio.Runner(*, debug=None, loop_factory=None)

run(coro, *, context=None)
Run a coroutine coro in the embedded loop

close()
Close the runner.

get_loop()
Return the event loop associated with the runner instance.

Example,

async def main():
    await asyncio.sleep(1)
    print('hey! whats up?')

with asyncio.Runner() as runner:
    runner.run(main())

Coroutine and Task API

Coroutine Function and Coroutine Object

# Define a coroutine function
async def my_function():
    print("Starting")
    await asyncio.sleep(1)
    print("Finished")
    return "Result"

# Call the function - this creates a coroutine object but doesn't run it yet
coro = my_function()
print(type(coro))  # <class 'coroutine'>

# To actually run it, you need to:
# Option 1: Await it (from within another coroutine)
result = await coro

# Option 2: Schedule it on the event loop
task = asyncio.create_task(coro)

# Option 3: Run it using asyncio.run()
result = asyncio.run(coro)

Key differences -

| Aspect              | Coroutine Function (`async def`)    | Coroutine Object (`func()`)               |
| ------------------- | ----------------------------------- | ----------------------------------------- |
|   Definition        | Declared with `async def`           | Created by calling a coroutine function   |
|   Nature            | A callable                          | An awaitable object                       |
|   Purpose           | Defines an asynchronous operation   | Represents the pending execution          |
|   Execution State   | Static (like a function definition) | Dynamic (has internal state)              |
|   Awaitable?        | ❌ Cannot be awaited                 | ✅ Can be awaited                          |
|   Scheduling        | Not scheduled on event loop         | Can be scheduled with `await` or a `Task` |
|                     |                                     |                                           |

Event Loop Life cycle Methods

# Explicit event loop management (when asyncio.run() isn't sufficient)
import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return "Done"

# Get the event loop
loop = asyncio.get_event_loop()

# Schedule coroutines
task = loop.create_task(my_coroutine())

# Run until all tasks complete
loop.run_until_complete(task)

# Get the result
result = task.result()
print(result)

# Close the loop when done
loop.close()

Creating Tasks API

import asyncio

async def fetch_data(id):
    print(f"Fetching data for {id}...")
    await asyncio.sleep(2)  # Simulating network request
    return f"Data for {id}"

async def main():
    # Create a task that runs in the background
    task = asyncio.create_task(fetch_data(123))

    # Do other work while the task runs concurrently
    print("Doing other work...")
    await asyncio.sleep(1)

    # Wait for the task to complete and get its result
    result = await task
    print(f"Task completed with result: {result}")

asyncio.run(main())

Task Cancellation and Handling Cancellation Exceptions

import asyncio

async def long_operation():
    try:
        print("Starting long operation...")
        await asyncio.sleep(10)  # Long operation
        return "Operation completed"
    except asyncio.CancelledError:
        print("Operation was cancelled, cleaning up...")
        # Perform any necessary cleanup
        raise  # Re-raise to propagate cancellation

async def main():
    # Create the task
    task = asyncio.create_task(long_operation())

    # Let it run for a bit
    await asyncio.sleep(2)

    # Cancel the task
    print("Cancelling task...")
    task.cancel()

    try:
        # Await the cancelled task
        await task
    except asyncio.CancelledError:
        print("Main: task was cancelled")

asyncio.run(main())

Task Groups API

import asyncio

async def process_item(item):
    print(f"Processing {item}...")
    await asyncio.sleep(item)  # Variable processing time
    return f"Processed {item}"

async def process_order(order):
    print(f"Processing {order}...")
    await asyncio.sleep(order)  # Variable processing order
    return f"Processed {order}"

async def process_bill(bill):
    print(f"Processing {bill}...")
    await asyncio.sleep(bill)  # Variable processing bill
    return f"Processed {bill}"

async def main():
    # Using TaskGroup to manage multiple tasks
    async with asyncio.TaskGroup() as tg:
        # Create multiple tasks that will run concurrently
        task1 = tg.create_task(process_item(1))
        task2 = tg.create_task(process_order(2))
        task3 = tg.create_task(process_bill(3))

        print("All tasks have been created and are running...")

    # When we exit the TaskGroup context, all tasks are awaited
    # If any task raises an exception, it will be propagated
    print("All tasks completed!")
    print(f"Results: {task1.result()}, {task2.result()}, {task3.result()}")

asyncio.run(main())

How await suspend execution and yeilds control back to event loop?

import asyncio
import time

async def task1():
    print(f"[{time.strftime('%H:%M:%S')}] Task 1 started")
    await asyncio.sleep(2)  # Suspend here, yield to event loop
    print(f"[{time.strftime('%H:%M:%S')}] Task 1 resumed after 2s")
    await asyncio.sleep(1)  # Suspend again
    print(f"[{time.strftime('%H:%M:%S')}] Task 1 completed")

async def task2():
    print(f"[{time.strftime('%H:%M:%S')}] Task 2 started")
    await asyncio.sleep(1)  # Suspend here, yield to event loop
    print(f"[{time.strftime('%H:%M:%S')}] Task 2 resumed after 1s")
    await asyncio.sleep(2)  # Suspend again
    print(f"[{time.strftime('%H:%M:%S')}] Task 2 completed")

async def main():
    # Create tasks to run concurrently
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())

    # Wait for both tasks to complete
    await t1
    await t2

asyncio.run(main())

Output :

Program starts
│
├─> Task 1 starts
│   └─> await sleep(2) - Suspends Task 1
│       │
├─> Task 2 starts      (Event loop switches to Task 2)
│   └─> await sleep(1) - Suspends Task 2
│       │
│       │              (Event loop waits, no active tasks)
│       │
│       └─> Task 2 resumes after 1s
│           └─> await sleep(2) - Suspends Task 2 again
│               │
│               └─> Task 1 resumes after 2s
│                   └─> await sleep(1) - Suspends Task 1 again
│                       │
│                       │              (Event loop waits)
│                       │
│                       └─> Task 1 completes
│                           │
│                           └─> Task 2 completes
│
Program ends

Understanding async withand async for Statements

Both async with and async for are extensions of their synchronous counterparts, designed to work with asynchronous code.
Both are syntactic sugar that make asynchronous code more readable and maintainable by matching the familiar synchronous patterns.

The async with statement is used with asynchronous context managers, allowing you to handle setup and teardown operations that involve awaitable objects.

Common for database connections, file operations, network connections

import asyncio

class AsyncResource:
    async def __aenter__(self):
        # Async setup code
        print("Acquiring resource...")
        await asyncio.sleep(1)  # Simulate async acquisition
        print("Resource acquired")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Async cleanup code
        print("Releasing resource...")
        await asyncio.sleep(0.5)  # Simulate async release
        print("Resource released")

    async def use_resource(self):
        print("Using the resource")
        await asyncio.sleep(0.5)

async def main():
    # Using async with to manage the resource
    async with AsyncResource() as resource:
        # This block is executed after __aenter__ completes
        await resource.use_resource()
        # When the block exits, __aexit__ is called automatically

    print("Done with the resource")

asyncio.run(main())

Acquiring resource...
Resource acquired
Using the resource
Releasing resource...
Resource released
Done with the resource

Another Example,

import asyncio
import aiofiles

async def read_file():
    async with aiofiles.open('data.txt', 'r') as file:
        # File is automatically closed when the block exits
        content = await file.read()
        return content

# Usage
asyncio.run(read_file())

The async for statement is used with asynchronous iterators, allowing you to iterate over items that are produced asynchronously.

Common for streaming APIs, paginated results, large dataset processing

import asyncio

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.counter = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.counter < self.limit:
            self.counter += 1
            # Simulate async work to produce next value
            await asyncio.sleep(0.5)
            return self.counter
        else:
            # Signal the end of iteration
            raise StopAsyncIteration

async def main():
    # Using async for to iterate over async iterator
    print("Starting iteration...")
    async for number in AsyncCounter(5):
        print(f"Got number: {number}")

    print("Iteration complete")

asyncio.run(main())

Starting iteration...
Got number: 1
Got number: 2
Got number: 3
Got number: 4
Got number: 5
Iteration complete

Another Example,

import asyncio
import aiohttp

async def fetch_data_stream():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data', timeout=30) as response:
            # Process the response as a stream
            async for chunk in response.content.iter_chunked(1024):
                # Process each chunk as it arrives
                process_chunk(chunk)

def process_chunk(chunk):
    # Do something with the data chunk
    print(f"Received {len(chunk)} bytes")

# Usage
asyncio.run(fetch_data_stream())

Understanding Futures in AsyncIO

Futures represent the eventual result of an asynchronous operation
They act as a placeholder until a result is available.
States:

Pending: Initial state, no result yet
Finished: Has a result or exception
Cancelled: Operation was cancelled

Key Methods:

set_result(result): Set the result of the future
set_exception(exception): Set an exception
result(): Get the result (raises exception if not done)
done(): Check if completed
add_done_callback(fn): Add a callback for when done

Comparison with Tasks:

Tasks are higher-level and manage coroutines
Tasks are automatically scheduled to run
Futures are lower-level, must be set manually
Every Task is a Future, but not every Future is a Task

Awaitable Objects ├── Coroutines (created from async def functions) ├── Tasks (manages coroutines in the event loop) └── Futures (low-level awaitable objects representing eventual results)

import asyncio

async def set_future_after_delay(future, delay, value):
    # Simulate work
    await asyncio.sleep(delay)
    # Set the result on the future
    future.set_result(value)
    print(f"Future set with value: {value}")

async def main():
    # Create a Future object
    future = asyncio.Future()

    # Schedule a task that will eventually set the future's result
    asyncio.create_task(set_future_after_delay(future, 2, "Future is complete!"))

    print("Waiting for future to complete...")

    # Await the future - this will pause here until the future has a result
    result = await future

    print(f"Received result: {result}")
    print(f"Future state: {future}")

asyncio.run(main())

Waiting for future to complete...
Future set with value: Future is complete!
Received result: Future is complete!
Future state: <Future finished result='Future is complete!'>

Cheat Sheet to save developers valuable time

✅ 1. Fundamental AsyncIO Concepts

| Concept                      | Purpose                         | Syntax                                      |
| ---------------------------- | ------------------------------- | ------------------------------------------- |
| Coroutine Function       | Declares async function         | `async def foo():`                          |
| Coroutine Object         | Returned when calling coroutine | `coro = foo()`                              |
| await                   | Suspend and yield control       | `await some_async_func()`                   |
| async with               | Async context manager           | `async with aiofile.open(...) as f:`        |
| async for                | Async iteration (e.g., streams) | `async for line in stream:`                 |
| Event Loop               | Orchestrates async tasks        | `asyncio.run(main())`                       |
| Manual Loop              | Custom control (rare)           | `loop = asyncio.get_event_loop()`           |
| asyncio.run()            | Start main coroutine            | `asyncio.run(main())`                       |
| asyncio.Runner() (3.11+) | Context-managed loop runner     | `with asyncio.Runner() as r: r.run(main())` |
| Event loop policy        | Platform-specific loop rules    | `asyncio.get_event_loop_policy()`           |

🧵 2. Concurrency Mechanisms

| Tool                  | Purpose                                  | Syntax                                                      |
| --------------------- | ---------------------------------------- | ----------------------------------------------------------- |
| Task              | Wraps coroutine for concurrent execution | `task = asyncio.create_task(coro())`                        |
| Cancel Task       | Stop task                                | `task.cancel()`                                             |
| TaskGroup (3.11+) | Manage related tasks together            | `async with asyncio.TaskGroup() as tg: tg.create_task(...)` |
| Future            | Placeholder for a result                 | `future = asyncio.Future()`                                 |
| Lock              | Mutual exclusion                         | `async with asyncio.Lock():`                                |
| Semaphore         | Limited concurrency                      | `async with asyncio.Semaphore(N):`                          |
| Event             | Wait for/set a signal                    | `await event.wait(); event.set()`                           |
| Queue            | Producer/Consumer coordination           | `await queue.put(x); x = await queue.get()`                 |

🔁 3. Concurrent Execution

| Tool                        | Purpose                               | Syntax                                          |
| --------------------------- | ------------------------------------- | ----------------------------------------------- |
| asyncio.gather()        | Run coroutines concurrently, wait all | `await asyncio.gather(task1, task2)`            |
| asyncio.wait()          | Wait for some/all tasks               | `done, pending = await asyncio.wait(tasks)`     |
| asyncio.as\_completed() | Yield tasks as they complete          | `async for res in asyncio.as_completed(tasks):` |
| asyncio.shield()        | Protect task from cancellation        | `await asyncio.shield(task)`                    |

🌐 4. Practical I/O

| Operation                 | Purpose                 | Syntax                                              |
| ------------------------- | ----------------------- | --------------------------------------------------- |
| open\_connection()   | Open TCP client         | `reader, writer = await asyncio.open_connection()`  |
| start\_server()       | Start TCP server        | `server = await asyncio.start_server(handler, ...)` |
| StreamReader/Writer   | Read/write via stream   | `data = await reader.read(); writer.write(b'data')` |
| File I/O (workaround) | Use thread executor     | `loop.run_in_executor(None, open, ...)`             |
| aiohttp/httpx         | Async HTTP client       | `await aiohttp.ClientSession().get(url)`            |
| asyncpg               | Async PostgreSQL client | `await asyncpg.connect(...)`                        |

⚠️ 5. Error Handling & Debugging

| Operation                | Purpose                | Syntax                                     |
| ------------------------ | ---------------------- | ------------------------------------------ |
| Try/Except in async  | Catch async exceptions | `try: await foo() except ...`              |
| Propagate Exceptions | Handled in task result | `result = await task`                      |
| Debugging            | Enable debug mode      | `asyncio.get_event_loop().set_debug(True)` |

In further post, focus would be to on solving problem with asyncio + FastAPI.

Happy Reading.... 🐍👩🏻‍💻🚀

DEV Community: Nazneen Mulani

Concurrency in Python

Fundamentals of Concurrency

Concurrency Models

Advantages and Limitations

asyncio — Asynchronous I/O