Reference library

Concurrency & performance

asyncio, threading, multiprocessing, and profiling-friendly performance patterns.

5 matches

How to Mock anyio.run Backends (asyncio vs trio) in Python

Demonstrates how to mock anyio.run to verify backend selection (asyncio or trio) without actually running the event loop.

anyio async testing

Python

import anyio
from unittest.mock import Mock, patch


async def fetch_data():
    await anyio.sleep(0.1)
    return {"data": 42}


def run_with_backend(backend: str):
    async def main():
        result = await fetch_data()
        print(f"[{backend}] Result: {result}")

    anyio.run(main, backend=backend)


if __nam…

14 0 Open

Concurrency & performance medium

How to Reduce Instance Memory with slots in Python

Demonstrates that classes with __slots__ use less memory per instance than regular classes because they skip the instance __dict__.

__slots__ memory performance

Python

class SlottedPoint:
    __slots__ = ('x', 'y', 'z')

    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z


class RegularPoint:
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z


if __name__ == "__main__":
    regular = RegularPoint(1, 2, 3)…

12 0 Open

Concurrency & performance medium

How to Run Blocking Code in an Executor with asyncio in Python

This code runs blocking functions concurrently without stalling the event loop by offloading them to thread pool executors via asyncio.

asyncio executor concurrency

Python

import asyncio
import time


def blocking_task(name: str, duration: float) -> str:
    """Simulate a blocking operation."""
    time.sleep(duration)
    return f"Finished {name} after {duration}s"


async def main() -> None:
    loop = asyncio.get_running_loop()
    results = await asyncio.gather(
        loop.run_in_…

14 0 Open

Concurrency & performance medium

How to Speed Up Downloads with ThreadPoolExecutor in Python

Compare sequential and thread-pool download loops to measure real speedup when I/O s bound.

threads concurrency performance

Python

import time
import threading
from concurrent.futures import ThreadPoolExecutor

def download_file(file_id):
    """Simulate fetching a file by sleeping briefly."""
    time.sleep(0.2)  # pretend network latency
    return f"file_{file_id}"

def sequential_downloads(num_files):
    """Process files one at a time."""
  …

13 0 Open

Concurrency & performance medium

asyncio Condition wait notify pattern in Python

Coordinate coroutines with asyncio.Condition: workers wait for notifications and the main task notifies one or all of them.

asyncio concurrency synchronization

Python

import asyncio


async def worker(condition, name):
    async with condition:
        print(f"{name} waiting...")
        await condition.wait()
        print(f"{name} notified!")


async def main():
    condition = asyncio.Condition()
    tasks = [asyncio.create_task(worker(condition, f"worker-{i}")) for i in range(3…

14 0 Open

Browse by section

Each section groups closely related Python snippets.