Reference library

Concurrency & performance

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

4 matches

How to Implement a Batch Requests Flush Interval in Python

A simple async batcher that accumulates items and flushes them either when a max batch size is reached or after a time-based flush interval.

asyncio batching concurrency

Python

import asyncio
from collections import deque

class Batcher:
    def __init__(self, flush_interval=0.5, max_batch=5):
        self.flush_interval = flush_interval
        self.max_batch = max_batch
        self.queue = deque()
        self.lock = asyncio.Lock()

    async def add(self, item):
        async with self.l…

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 easy

How to Use Array Typecodes for Compact Numeric Storage in Python

This code demonstrates how to use the `array` module with typecodes to store integers, floats, and bytes in a memory-efficient way compared to standard Python lists.

array memory performance

Python

from array import array

def demonstrate_array_types():
    # Compact integer arrays
    small_ints = array('i', [1, 2, 3, 4, 5])
    unsigned_ints = array('I', [10, 20, 30])
    
    # Floating point arrays
    floats = array('f', [1.5, 2.5, 3.5])
    doubles = array('d', [1.123456789, 2.987654321])
    
    # Charac…

15 0 Open

Concurrency & performance medium

Merge K Sorted Lists in Python with heapq

Merge k sorted lists into one sorted list in O(N log k) time using a min-heap of current elements.

heapq merge sorted-lists

Python

import heapq

def merge_k_sorted_lists(lists):
    heap = []
    for i, lst in enumerate(lists):
        if lst:  # only push non-empty lists
            heapq.heappush(heap, (lst[0], i, 0))
    result = []
    while heap:
        val, list_idx, elem_idx = heapq.heappop(heap)
        result.append(val)
        if elem…

13 0 Open

Browse by section

Each section groups closely related Python snippets.