Reference library

Concurrency & performance

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

3 matches

How to Mock asyncio.open_connection in Python

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

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 Send and Receive Messages Between Processes with multiprocessing.Pipe in Python

Use multiprocessing.Pipe to create a two-way connection between two processes, send a message from parent to child, and receive a reply back.

multiprocessing pipe interprocess-communication

Python

import multiprocessing


def child_process(conn):
    """Receive from parent and send back a response."""
    message = conn.recv()
    print(f"Child received: {message}")
    conn.send("Hello from child!")


if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()

    process = multiprocessing…

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

Browse by section

Each section groups closely related Python snippets.