Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to Use fcntl for Exclusive File Locking in Python
This code demonstrates how to acquire an exclusive advisory lock on a file using fcntl.flock with a non-blocking flag, simulate work, then release the lock.
import fcntl
import os
import tempfile
import time
def acquire_exclusive_lock(filepath):
fd = os.open(filepath, os.O_RDWR | os.O_CREAT)
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
print(f"Exclusive lock acquired on {filepath}")
time.sleep(1) # Simulate work while holding the l…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…Find Broken Image References Across a Website in Python
Crawl internal pages of a website, collect all image source URLs, then check each with HEAD requests to report any that return HTTP 4xx or connection errors.
import requests
from urllib.parse import urljoin, urlparse
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor, as_completed
def find_all_links(base_url, max_pages=50):
visited, to_visit = set(), {base_url}
while to_visit and len(visited) < max_pages:
url = to_visit.pop()
…How to Build a Cryptocurrency Price Tracker in Python
A continuous Python script that fetches real-time cryptocurrency prices from the CoinGecko API and displays them on a loop.
import requests
import time
def get_crypto_prices(coin_ids=["bitcoin", "ethereum", "solana"]):
url = "https://api.coingecko.com/api/v3/simple/price"
params = {
"ids": ",".join(coin_ids),
"vs_currencies": "usd"
}
try:
response = requests.get(url, params=params, timeout=10)
…Graceful Shutdown Executor Context Manager in Python
A context manager that starts a background thread and ensures it stops gracefully on exit, handling timeouts and exceptions.
import signal
import threading
import time
from contextlib import contextmanager
@contextmanager
def graceful_shutdown_executor(timeout=5.0):
"""Context manager that runs a task and gracefully stops it on timeout or exception."""
stop_event = threading.Event()
def task():
print("Task started")
…How to Build a Producer-Consumer Pattern with asyncio.Queue in Python
This code implements a classic producer-consumer pattern using asyncio.Queue to coordinate one producer task that generates items and two consumer tasks that process them concurrently, with a sentinel value to signal completion.
import asyncio
import random
async def producer(queue, item_count):
for i in range(item_count):
item = random.randint(1, 100)
await queue.put(item)
print(f"Produced: {item}")
await asyncio.sleep(0.1)
await queue.put(None) # Sentinel to signal end
async def consumer(queue, n…How to Cancel an asyncio Task with Graceful Cleanup in Python
Cancel a running asyncio task, handle the cancellation signal inside a worker coroutine to perform cleanup, then re-raise so the cancellation propagates correctly.
import asyncio
async def worker(name: str, sleep: float) -> None:
try:
print(f"{name}: starting")
await asyncio.sleep(sleep)
print(f"{name}: completed")
except asyncio.CancelledError:
print(f"{name}: cancelled, cleaning up...")
await asyncio.sleep(0.2) # Simulate clea…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.
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…How to Implement a Token Bucket Rate Limiter with asyncio in Python
This code implements a thread-safe token bucket rate limiter for asyncio, allowing you to limit the rate of async tasks or API calls.
import asyncio
import time
class TokenBucket:
def __init__(self, rate_per_second, capacity):
self.rate = rate_per_second
self.capacity = capacity
self.tokens = capacity
self.last_refill = time.monotonic()
self.lock = asyncio.Lock()
async def acquire(self):
asy…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 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 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 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 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 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 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 Use asyncio Lock to Protect a Shared Counter in Python
This code demonstrates how to use an asyncio.Lock to safely increment a shared counter from multiple concurrent coroutines.
import asyncio
async def increment(counter, lock, increments):
for _ in range(increments):
async with lock:
counter[0] += 1
async def main():
counter = [0]
lock = asyncio.Lock()
tasks = [
increment(counter, lock, 1000)
for _ in range(5)
]
await asyncio.gath…How to Use multiprocessing Pool map and starmap in Python
Parallelize functions over iterables with Pool.map, and unpack multiple arguments via Pool.starmap.
from multiprocessing import Pool
def square(x):
return x * x
def add_and_multiply(a, b, c):
return (a + b) * c
if __name__ == "__main__":
numbers = [1, 2, 3, 4, 5]
with Pool(processes=2) as pool:
squares = pool.map(square, numbers)
print(f"squares: {squares}")
starmap_arg…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.