Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
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_…How to Parse JSON Files in Parallel with Python ThreadPoolExecutor
Load and transform JSON records from multiple files concurrently using ThreadPoolExecutor for faster I/O-bound parsing.
import time
from concurrent.futures import ThreadPoolExecutor
import json
def load_json_file(path):
with open(path, 'r') as f:
return json.load(f)
def transform_record(record):
record['full_name'] = f"{record.pop('first_name', '')} {record.pop('last_name', '')}".strip()
record['score'] = int(reco…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.
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(…How to Profile CPU Hot Path in Python with cProfile and sort_stats cumtime
Profile a Python function's CPU usage by running cProfile, sorting stats by cumulative time, and printing a readable report to stdout.
import cProfile
import pstats
import io
def slow_function():
total = 0
for i in range(100_000):
total += i * i
return total
def fast_function():
return sum(i for i in range(100))
def main():
slow_function()
fast_function()
if __name__ == "__main__":
profiler = cProfile.Profi…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)…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_…How to Run Coroutines Concurrently with asyncio.gather in Python
Run multiple async coroutines concurrently and collect their results in the order they were passed.
import asyncio
async def fetch_data(name: str, delay: float) -> str:
"""Simulate an async operation (e.g., API call) with a delay."""
await asyncio.sleep(delay)
return f"{name} data (after {delay}s)"
async def main() -> None:
"""Run multiple coroutines concurrently with asyncio.gather."""
resul…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.
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…How to Share Memory Between Processes in Python with multiprocessing.Value and Array
Share a numeric value and a list-like array across multiple Python processes using multiprocessing.Value and multiprocessing.Array, with each process modifying the same memory.
import multiprocessing
def worker(shared_value, shared_array, index):
shared_value.value += 10
shared_array[index] = shared_array[index] * 2
if __name__ == "__main__":
shared_value = multiprocessing.Value("i", 5)
shared_array = multiprocessing.Array("i", [1, 2, 3, 4, 5])
processes = []
for i…How to Share a Dict and List Between Processes with multiprocessing Manager in Python
This code demonstrates how to share a dictionary and a list between multiple processes using multiprocessing.Manager, enabling safe concurrent updates.
import multiprocessing as mp
def worker(shared_dict, shared_list, name):
shared_dict[name] = name.upper()
shared_list.append(name)
print(f"{name} added to shared structures")
def main():
with mp.Manager() as manager:
shared_dict = manager.dict()
shared_list = manager.list()
…How to Share a Queue Between Processes in Python
Use multiprocessing.Queue to pass work from a producer process to multiple consumer processes, coordinating with a sentinel stop message.
import multiprocessing
import time
def producer(queue, items):
for item in items:
queue.put(item)
time.sleep(0.1)
queue.put("STOP")
def consumer(queue, name):
while True:
item = queue.get()
if item == "STOP":
break
print(f"{name} processed: {item}")
…How to Signal asyncio Workers to Stop with an Event in Python
Use an asyncio.Event to coordinate graceful shutdown of multiple concurrent worker tasks in Python.
import asyncio
import random
async def worker(name, stop_event):
while not stop_event.is_set():
await asyncio.sleep(random.uniform(0.1, 0.5))
print(f"Worker {name} processing...")
print(f"Worker {name} stopped.")
async def main():
stop_event = asyncio.Event()
workers = [asyncio.create…How to Speed Up Data Filtering with Python ThreadPoolExecutor
This code compares sequential filtering of even numbers with a threaded version using ThreadPoolExecutor, showing a measurable speedup for I/O-bound work.
import time
from concurrent.futures import ThreadPoolExecutor
import random
def is_even(number):
time.sleep(0.001) # simulate work
return number % 2 == 0
def filter_even_sequential(numbers):
return [n for n in numbers if is_even(n)]
def filter_even_threaded(numbers):
with ThreadPoolExecutor(max_…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."""
…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.
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…How to Time Code Performance with timeit in Python
Benchmark two implementations of the same logic using Python's timeit module and compare their execution speeds.
import timeit
# Implementation 1: Using a list comprehension
def list_comprehension_squares(n):
return [i ** 2 for i in range(n)]
# Implementation 2: Using a for loop with append
def loop_squares(n):
result = []
for i in range(n):
result.append(i ** 2)
return result
if __name__ == "__main__"…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.
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…How to Use ProcessPoolExecutor for CPU Parallel Map in Python
Run a function over a sequence of inputs in parallel across multiple CPU cores with ProcessPoolExecutor.map.
from concurrent.futures import ProcessPoolExecutor
import math
def compute_square(num):
return num * num
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
if __name__ == "__main__":
numbers = rang…How to Use Thread Pool Executor map for IO-Bound Tasks in Python
Run multiple I/O-bound tasks concurrently with ThreadPoolExecutor map and collect their results in order.
import time
from concurrent.futures import ThreadPoolExecutor
def io_bound_task(task_id: int) -> str:
time.sleep(0.2) # mock I/O wait
return f"Task {task_id} completed"
def main() -> None:
task_ids = [1, 2, 3, 4, 5]
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.…How to Use ThreadPoolExecutor and ProcessPoolExecutor in Python
Compares ThreadPoolExecutor and ProcessPoolExecutor by running CPU-bound and I/O-tolerant tasks over a large list, printing elapsed times and first results.
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import math
numbers = list(range(1, 1000001))
def compute_square(n):
return n * n
def compute_sqrt(n):
return math.sqrt(n)
def run_executor(executor, func, data):
start = time.perf_counter()
results = list(executo…How to Use ThreadPoolExecutor for Concurrent Tasks in Python
Compare sequential execution with ThreadPoolExecutor for I/O-bound tasks, measuring speedup and timing with perf_counter.
import time
import threading
from concurrent.futures import ThreadPoolExecutor
def fetch_data(index):
"""Simulate a synchronous data fetch."""
time.sleep(0.1)
return f"data-{index}"
def run_sequential(total=10):
"""Run tasks one after another."""
start = time.perf_counter()
results = [fetch…How to Use ThreadPoolExecutor in Python for Parallel Processing
Use ThreadPoolExecutor with executor.map to run a function over many inputs concurrently and collect ordered results.
def worker(item):
return item * item
if __name__ == "__main__":
from concurrent.futures import ThreadPoolExecutor
numbers = list(range(1, 11))
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(worker, numbers))
print("Input: ", numbers)
print("Results:", …How to Use ThreadPoolExecutor.submit() in Python
Exécute des fonctions en parallèle avec ThreadPoolExecutor.submit(), récupère les résultats avec future.result(), et traite plusieurs tâches simultanément en Python standard.
from concurrent.futures import ThreadPoolExecutor
import time
def square(n):
time.sleep(0.1) # Simulate work
return n * n
if __name__ == "__main__":
with ThreadPoolExecutor(max_workers=3) as executor:
future = executor.submit(square, 5)
result = future.result()
print(f"Result: {r…How to Use a Bounded Buffer with threading.Condition in Python
Implement a thread-safe bounded buffer using threading.Condition and show a producer–consumer example with exact output.
import threading
import time
import random
class BoundedBuffer:
def __init__(self, capacity):
self.capacity = capacity
self.buffer = []
self.condition = threading.Condition()
def put(self, item):
with self.condition:
while len(self.buffer) >= self.capacity:
…Browse by section
Each section groups closely related Python snippets.
Guide: free Python code samples library
Copy-ready Python snippets for learners and developers
PythonSkillset code samples are short, focused examples organised by topic and difficulty. Every snippet is server-rendered HTML — readable by search engines and easy to copy. Open any sample, read the notes, copy the code, then press Try in editor to run it in the browser with Pyodide.
How to use this library
- Pick a topic section — strings, lists, files, functions, and more
- Open a sample, read How it works, and copy the code block
- Run it in the IDE, tweak values, then take a related quiz or tutorial lesson
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.