Concurrency &amp; performance

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…

__slots__ memory performance

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__.

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)…

11 0 Open

asyncio executor concurrency

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.

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

asyncio event loop entry point

How to Run an Async Main with asyncio.run in Python

Show the canonical entry point for an asyncio program: define an async main, then launch it with asyncio.run.

import asyncio


async def main():
    print("Hello from async main")
    await asyncio.sleep(0.1)
    print("Done")


if __name__ == "__main__":
    asyncio.run(main())

16 0 Open

threads concurrency performance

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.

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."""
  …

uvloop asyncio event-loop

How to Use uvloop Faster Event Loop

Install uvloop at startup to replace asyncio's default event loop with a faster libuv-based one, with a graceful fallback when it's unavailable.

import asyncio
try:
    import uvloop
    uvloop.install()
    USING_UVLOOP = True
except ImportError:
    USING_UVLOOP = False


async def fetch_data(index):
    await asyncio.sleep(0.01)
    return f"data-{index}"


async def main():
    tasks = [fetch_data(i) for i in range(10)]
    results = await asyncio.gather(*…

14 0 Open

asyncio timeout concurrency

How to set a timeout with asyncio.wait_for in Python

Use asyncio.wait_for to bound an async function with a timeout, catching TimeoutError when it exceeds the limit.

import asyncio

async def slow_task():
    await asyncio.sleep(3)
    return "finished"

async def main():
    try:
        result = await asyncio.wait_for(slow_task(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out")

if __name__ == "__main__":
    asyncio.run(main())

asyncio async concurrency

Run Background Tasks with asyncio.create_task in Python

Create background tasks in an asyncio event loop with asyncio.create_task and run them concurrently using asyncio.gather.

import asyncio
import time

async def background_worker(name, duration):
    """Simulates a long-running background task."""
    print(f"{name} started at t={time.monotonic():.1f}")
    await asyncio.sleep(duration)
    print(f"{name} finished at t={time.monotonic():.1f}")

async def main():
    print(f"Main starting …

asyncio concurrency synchronization

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.

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

asyncio concurrency scheduling

asyncio sleep cooperative scheduling demo in Python

This demo shows how asyncio.sleep yields control between concurrent tasks, letting multiple workers interleave their ticks.