Python Concurrency

Understanding the Concurrency Models

Python offers three approaches:

threading

Use threads when you’re doing I/O-bound work with blocking code or third-party libraries.

import threading
import time
import requests

def fetch_url(url, results, index):
    response = requests.get(url)
    results[index] = response.status_code

urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

results = [None] * len(urls)
threads = []

for i, url in enumerate(urls):
    t = threading.Thread(target=fetch_url, args=(url, results, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(results)   # [200, 200, 200] — all fetched concurrently

Thread Safety with Lock

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    with lock:
        counter += 1

threads = [threading.Thread(target=increment) for _ in range(1000)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # 1000 — safe with lock

Thread-Local Storage

local_data = threading.local()

def process():
    local_data.user_id = threading.current_thread().name
    # Each thread has its own local_data.user_id

multiprocessing

Use processes for CPU-bound work — each process gets its own Python interpreter and memory.

from multiprocessing import Pool
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

numbers = list(range(100_000, 100_100))

# Sequential — uses one CPU core
primes_seq = [n for n in numbers if is_prime(n)]

# Parallel — uses all CPU cores
with Pool() as pool:
    results = pool.map(is_prime, numbers)
    primes_par = [n for n, is_p in zip(numbers, results) if is_p]

Process Communication with Queue

from multiprocessing import Process, Queue

def worker(q, items):
    for item in items:
        q.put(item * 2)

q = Queue()
p = Process(target=worker, args=(q, [1, 2, 3, 4]))
p.start()
p.join()

while not q.empty():
    print(q.get())   # 2, 4, 6, 8

concurrent.futures

The high-level interface for both threads and processes — simpler and more Pythonic.

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import requests

urls = ["https://httpbin.org/get"] * 5

# ThreadPoolExecutor for I/O-bound
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(requests.get, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            print(f"{url}: {result.status_code}")
        except Exception as e:
            print(f"{url} failed: {e}")

# ProcessPoolExecutor for CPU-bound
def heavy_compute(n):
    return sum(i**2 for i in range(n))

with ProcessPoolExecutor() as executor:
    results = list(executor.map(heavy_compute, [100_000] * 8))

asyncio

The gold standard for high-concurrency I/O — handles thousands of connections in a single thread.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3",
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

posts = asyncio.run(main())

async / await Patterns

import asyncio

async def slow_operation(name, delay):
    print(f"{name}: starting")
    await asyncio.sleep(delay)   # yields control to event loop
    print(f"{name}: done after {delay}s")
    return name

async def main():
    # Sequential — total 3 seconds
    await slow_operation("A", 1)
    await slow_operation("B", 2)

    # Concurrent — total ~2 seconds
    results = await asyncio.gather(
        slow_operation("A", 1),
        slow_operation("B", 2),
    )

    # With timeout
    try:
        result = await asyncio.wait_for(slow_operation("C", 10), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

asyncio.run(main())

Async Context Managers and Iterators

import asyncio

class AsyncDatabase:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, *args):
        await self.close()

    async def fetch_rows(self, query):
        async for row in self.execute(query):
            yield row

async def main():
    async with AsyncDatabase() as db:
        async for row in db.fetch_rows("SELECT * FROM users"):
            print(row)

asyncio.TaskGroup (Python 3.11+)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("https://api.example.com/users"))
        task2 = tg.create_task(fetch_data("https://api.example.com/posts"))
    # All tasks complete before this line
    users = task1.result()
    posts = task2.result()

TaskGroup is preferred over gather because it cancels remaining tasks on failure.

Choosing the Right Tool

# I/O-bound + async-aware library → asyncio
async def handler(request):
    data = await db.fetch(request.user_id)
    return data

# I/O-bound + blocking library → ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=10) as ex:
    results = list(ex.map(blocking_http_call, urls))

# CPU-bound → ProcessPoolExecutor or multiprocessing
with ProcessPoolExecutor() as ex:
    results = list(ex.map(cpu_heavy_function, data))

# Mix async + blocking code → run_in_executor
async def mixed():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_function, arg)