Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to parallel map embeddings with a thread pool in Python
Run embedding computations in parallel using ThreadPoolExecutor, collect results into a dict keyed by the original item.
import threading
from concurrent.futures import ThreadPoolExecutor
import time
def compute_embedding(item: int) -> tuple[int, int]:
time.sleep(0.05) # Simulate embedding work
return item, item * 10
def parallel_map_embed(items, max_workers=3):
results = {}
with ThreadPoolExecutor(max_workers=max_w…How to Ping Multiple Hosts in Parallel with Python ThreadPoolExecutor
A parallel host-pinging script using ThreadPoolExecutor and subprocess to check connectivity across multiple addresses concurrently.
import subprocess
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
HOSTS = [
"google.com",
"github.com",
"stackoverflow.com",
"nonexistent.invalid",
"localhost",
]
def ping_host(host: str) -> str:
"""Ping a single host and return a status string."""
result = subp…Parallel Extract Multiple Sources with Threads in Python
Extract data from multiple sources in parallel using ThreadPoolExecutor and verify results match sequential processing.
import threading
from concurrent.futures import ThreadPoolExecutor
def extract_from_source(source):
"""Simulate extracting data from a source."""
return f"Data from {source}"
def main():
sources = ["source_a", "source_b", "source_c", "source_d"]
# Sequential extraction for comparison
sequent…How to Convert Data in Parallel with ThreadPoolExecutor in Python
This example demonstrates converting a list of items in parallel using ThreadPoolExecutor, showing performance gains over serial processing.
import time
from concurrent.futures import ThreadPoolExecutor
def convert_data(item):
"""Simulate a CPU/IO-bound conversion task."""
time.sleep(0.05) # simulate work
return item.upper()
if __name__ == "__main__":
items = [f"item_{i}" for i in range(20)]
start = time.perf_counter()
serial_…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 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 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 as_completed to Process Futures in Order of Completion
Submit multiple tasks to a ThreadPoolExecutor and process each result as soon as it finishes using as_completed.
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def fetch_data(item_id):
time.sleep(1)
return f"item-{item_id}"
def main():
with ThreadPoolExecutor(max_workers=3) as executor:
future_map = {executor.submit(fetch_data, i): i for i in range(1, 6)}
for future in…How to Validate Data with ThreadPoolExecutor in Python
This code shows how to validate a list of numbers concurrently using ThreadPoolExecutor, dramatically speeding up slow validation tasks by running them in parallel threads.
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
@dataclass
class Result:
is_valid: bool
value: int
def validate(value: int) -> Result:
time.sleep(0.1) # simulate slow validation (API call, DB check)
return Result(is_valid=0 < value < 100, value=value…How to use ThreadPoolExecutor for concurrent tasks in Python
Run blocking functions in parallel with ThreadPoolExecutor and as_completed, cutting total runtime from 5 sequential sleeps to about 1 second.
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def fetch_data(item):
"""Simulate a slow operation with a fixed delay."""
time.sleep(0.2)
return item * 2
def main():
items = [1, 2, 3, 4, 5]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=3) as ex…Thread Pool Map for IO Bound Tasks in Python
Run IO-bound mock tasks concurrently with ThreadPoolExecutor.map and measure total elapsed time in Python.
import concurrent.futures
import time
from pathlib import Path
def mock_io_task(filename):
"""Simulate an IO-bound task by creating a small file and measuring its latency."""
path = Path(filename)
path.write_text("data")
time.sleep(0.1) # Simulate slow disk/network
return f"{filename} written in …How to Limit Concurrent Requests with a Semaphore in Python
Use threading.Semaphore with a ThreadPoolExecutor to cap how many worker threads run simultaneously, preventing resource overload.
import threading
import time
from concurrent.futures import ThreadPoolExecutor
def worker(name, semaphore, results):
with semaphore:
results.append(f"start {name}")
time.sleep(0.5) # simulate async work
results.append(f"done {name}")
def main():
sem = threading.Semaphore(2) # max 2 …Implement Bulkhead Thread Pool Isolation in Python
Create isolated thread pools with a bulkhead pattern to protect different services from cascading failures.
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class Bulkhead:
"""Simple bulkhead isolation: separate thread pools for different tasks."""
def __init__(self, max_workers):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.active = …Bulkhead Thread Pool per Service Mock in Python
Simulates a bulkhead pattern with per-service thread pools and semaphore-based rejection to isolate failures between dependent services.
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class ServiceBulkhead:
def __init__(self, name, max_threads, max_queue):
self.name = name
self.executor = ThreadPoolExecutor(max_workers=max_threads)
self.semaphore = threading.Semaphore(max_thread…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.