Reference library

Concurrency & performance

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

4 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 Pause and Resume Threads with threading.Event in Python

Use threading.Event to pause and resume worker threads in Python, controlling execution flow with set and clear methods.

threading events concurrency

Python

import threading
import time

workers = []

def worker(name, event):
    for i in range(10):
        event.wait()
        print(f"{name} step {i}")
        time.sleep(0.1)

def pause_worker(name):
    global pause_event
    for w in workers:
        if w.name == name:
            pause_event.clear()
            print(…

10 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_…

13 0 Open

Concurrency & performance medium

How to Use a Weakref Cache to Avoid Memory Leaks in Python

This code demonstrates building a value cache with weakref.WeakValueDictionary so objects can be garbage collected when no longer referenced, preventing memory leaks.

weakref caching memory

Python

import weakref
import gc


class ExpensiveObject:
    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return f"ExpensiveObject('{self.name}')"


class ObjectCache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

    def get_or_create(self, name):
       …

13 0 Open

Browse by section

Each section groups closely related Python snippets.