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…

13 0 Open

Concurrency & performance medium

How to Mock asyncio.open_connection in Python

Mock asyncio.open_connection with AsyncMock to test async code without a real network connection.

asyncio testing mocking

Python

import asyncio
from unittest.mock import AsyncMock, patch


async def fetch_data(reader: asyncio.StreamReader) -> str:
    data = await reader.readline()
    return data.decode().strip()


async def main() -> None:
    # Mock asyncio.open_connection to simulate a server response
    mock_reader = AsyncMock()
    mock_…

14 0 Open

Concurrency & performance easy

How to Test HTTPX Async Client Pool Reuse with Mocks in Python

Mock an httpx.AsyncClient to verify connection pool reuse by asserting GET calls share a single client instance across concurrent async requests.

httpx async-await mock

Python

import asyncio
import httpx
from unittest.mock import AsyncMock, patch, Mock

async def fetch_with_pool(client, url, n_reuses=3):
    results = []
    for i in range(n_reuses):
        resp = await client.get(url)
        results.append(resp.status_code)
        await asyncio.sleep(0)  # yield to loop to mimic real us…

13 0 Open

Concurrency & performance medium

Mocking Trio's open_nursery and spawn with asyncio.TaskGroup

Show how to mock Trio's nursery pattern using Python's asyncio.TaskGroup to simulate task spawning and completion.

asyncio taskgroup concurrency

Python

import asyncio

class MockSpawner:
    async def spawn(self, nursery):
        print("Spawning mock task...")
        await asyncio.sleep(1)
        print("Mock task completed")

async def open_nursery():
    async with asyncio.TaskGroup() as nursery:
        mock = MockSpawner()
        nursery.create_task(mock.spawn…

14 0 Open

Concurrency & performance medium

Thread Pool Map for IO Bound Tasks in Python

Run IO-bound mock tasks concurrently with ThreadPoolExecutor.map and measure total elapsed time in Python.

threading concurrency threadpoolexecutor

Python

import concurrent.futures
import time
from pathlib import Path

def mock_io_task(filename):
    """Simulate an IO-bound task by creating a small file and measuring its latency."""
    path = Path(filename)
    path.write_text("data")
    time.sleep(0.1)  # Simulate slow disk/network
    return f"{filename} written in …

14 0 Open

Browse by section

Each section groups closely related Python snippets.