Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
Build a Generator Pipeline in Python: Filter Then Map
Create a lazy data pipeline by chaining generator functions that read, filter, map, and write data step by step.
def read_data():
return ["a", "bb", "ccc", "dd", "eeeee", "f"]
def filter_short(words):
return (word for word in words if len(word) >= 2)
def map_to_upper(words):
return (word.upper() for word in words)
def write_data(words):
for word in words:
print(word)
if __name__ == "__main__":
…How to Build a Backpressure Generator Pause Producer Demo in Python
Demonstrates a producer–consumer pattern with a fixed-size buffer that pauses production when full, simulating backpressure.
import time
import collections
def producer(buffer, max_size, items):
"""Adds items to the buffer until full, then pauses."""
for item in items:
while len(buffer) >= max_size:
print(f"Buffer full ({len(buffer)}/{max_size}) — producer paused")
time.sleep(0.1)
buffer.appe…How to Create a Generator Context Manager in Python with contextlib
Create a custom context manager with the @contextlib.contextmanager decorator to manage resources using a generator function.
import contextlib
@contextlib.contextmanager
def temporary_directory():
"""Yield a string and clean up after the block exits."""
print("Creating temp directory...")
dir_name = "/tmp/example"
try:
yield dir_name
finally:
print(f"Removing {dir_name}...")
if __name__ == "__main__":
…How to Generate Primes with a Generator in Python
Generate prime numbers up to a limit using the Sieve of Eratosthenes wrapped in a generator expression for lazy evaluation.
def prime_generator(limit):
sieve = [True] * (limit + 1)
sieve[0] = sieve[1] = False
for i in range(2, int(limit ** 0.5) + 1):
if sieve[i]:
for j in range(i * i, limit + 1, i):
sieve[j] = False
return (num for num, is_prime in enumerate(sieve) if is_prime)
if __n…How to Send Values into a Python Generator Coroutine
Use the .send() method to pass values into a running generator coroutine and capture them.
def coroutine():
received = []
while True:
value = yield
received.append(value)
print(f"Coroutine received: {value}")
if value == "stop":
break
return received
if __name__ == "__main__":
gen = coroutine()
next(gen) # Prime the generator
gen.send("he…How to Throw an Exception into a Python Generator
This code demonstrates how to use the .throw() method on a generator to inject an exception at its current yield point and let it recover gracefully.
def demo_throw_into_generator():
"""Demonstrate throwing an exception into a running generator."""
def counter():
"""Generator that counts until interrupted."""
try:
i = 0
while True:
yield i
i += 1
except ValueError as e:
…How to filter a generator with a predicate function in Python
This code defines a generator function that yields only items from an iterable that satisfy a given predicate, then tests it with even and positive number filters.
def filter_gen(predicate, iterable):
for item in iterable:
if predicate(item):
yield item
def is_even(num):
return num % 2 == 0
def is_positive(num):
return num > 0
if __name__ == "__main__":
numbers = range(-5, 10)
even_numbers = list(filter_gen(is_even, numbers))
p…How to stream parse JSON arrays in Python
This code demonstrates two generators: one that streams a JSON array as individual chunks, and another that incrementally parses those chunks into Python objects using json.JSONDecoder.
import json
def json_array_stream(items):
"""Generator that yields JSON-encoded values one at a time."""
yield "["
for i, item in enumerate(items):
if i > 0:
yield ","
yield json.dumps(item)
yield "]"
def parse_json_stream(stream):
"""Consumes a stream of JSON fragme…Merge Sorted Iterators with a Heap Generator in Python
Merge multiple sorted iterators into a single sorted stream using a heap and generator, yielding values lazily in order.
import heapq
def merge_sorted_iterators(*iterators):
heap = []
for idx, iterator in enumerate(iterators):
try:
value = next(iterator)
heapq.heappush(heap, (value, idx, iterator))
except StopIteration:
continue
while heap:
value, idx, iterator = …Circuit Breaker Pattern in Python for LLM API Calls
Implements a circuit breaker class that wraps LLM client calls to fail fast when the service is degrading, then recover automatically after a timeout.
import time
class CircuitBreaker:
def __init__(self, failure_threshold=3, recovery_timeout=5):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.state = "closed"
self.last_failure_time = None
def call(self, …How to Build a Data Helper for LLM Prompts in Python
A beginner-friendly helper class that flattens nested dictionaries, formats prompt templates, and safely parses JSON for AI/LLM pipelines.
import json
from typing import Any, Dict, List, Optional
class DataHelper:
"""Simple helper class for working with data in AI/LLM pipelines."""
def __init__(self, data: Optional[Dict[str, Any]] = None) -> None:
self.data = data or {}
def flatten(self, prefix: str = "") -> Dict[str, Any]…How to Detect Prompt Injection in Python
Implements a regex-based heuristic in Python to flag common prompt injection attempts before sending input to an LLM.
import re
def contains_prompt_injection(user_input: str) -> bool:
# Directives to ignore previous instructions or act as system
ignore_patterns = [
r"\bignore\s+(all\s+)?previous\s+instructions\b",
r"\bdisregard\s+(all\s+)?previous\s+instructions\b",
r"\bdon'?t\s+follow\s+(any\s+)?inst…How to Repair Malformed JSON Braces Heuristically in Python
Heuristically fix malformed JSON by balancing braces and quotes, using a stack-based approach to add missing closing characters.
import json
import re
def repair_json(text: str) -> str:
"""Heuristically repair malformed JSON by balancing braces and quotes."""
# Trim whitespace and handle leading/trailing garbage
text = text.strip()
# Remove common non-JSON decorations
text = re.sub(r'^(How to Retry LLM Calls on Rate Limit Errors in Python
Implement a retry mechanism with exponential backoff for LLM API calls that raises a custom RateLimitError, using a mock function to demonstrate the pattern.
import time
import random
def mock_llm_call():
"""Simulates an LLM API call that may raise a rate limit error."""
if random.random() < 0.4: # 40% chance of rate limit
raise RateLimitError("Rate limit exceeded. Try again later.")
return {"response": "Hello world from mock LLM"}
class RateLimitE…How to cache embeddings with a Python dict to avoid recomputation
Caches embeddings computed from text in a dictionary keyed by SHA-256 hash, returning cached results for repeated calls.
import hashlib
import time
class EmbeddingCache:
def __init__(self):
self.cache = {}
def _hash_text(self, text):
return hashlib.sha256(text.encode()).hexdigest()
def get_embedding(self, text, compute_func):
key = self._hash_text(text)
if key not in self.cache:
…How to implement exponential backoff for LLM API calls in Python
A decorator that retries flaky LLM API calls with exponential delay, using a mock client to demonstrate the pattern.
import time
import random
class MockLLM:
def call(self, prompt):
if random.random() < 0.7: # 70% chance of transient failure
raise ConnectionError("API unavailable")
return f"LLM response for: {prompt}"
def with_exponential_backoff(max_retries=5, base_delay=0.1):
def decorator(fu…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…Parse ReAct Logs into Thought Action Observation Steps in Python
Parse a ReAct agent's textual log into structured steps with thought, action, and observation using regex and named tuples.
import re
from collections import namedtuple
ReActStep = namedtuple("ReActStep", ["thought", "action", "observation"])
def parse_react_log(log: str) -> list[ReActStep]:
"""Parse a ReAct log into structured thought/action/observation steps."""
pattern = re.compile(
r"Thought:\s*(?P<thought>.+?)\s*"
…Track GitHub Repository Growth in Python
A Python dashboard that fetches and displays GitHub repository statistics including stars, forks, creation date, and recent star activity using the GitHub API.
import requests
import json
from datetime import datetime, timedelta
def track_repo_growth(owner, repo):
url = f"https://api.github.com/repos/{owner}/{repo}"
headers = {"Accept": "application/vnd.github.v3+json"}
response = requests.get(url, headers=headers)
data = response.json()
name = data…Automatically Clean Temporary Files from Applications Using Python
A Python script that safely deletes temporary files from common application temp directories across Windows, Linux, and macOS, tracking cleaned count and disk space.
import os
import shutil
import tempfile
import platform
def clean_application_temp_files():
"""Delete common temporary file locations safely."""
system = platform.system()
temp_dirs = []
if system == "Windows":
temp_dirs.extend([
os.path.join(os.getenv("LOCALAPPDATA"), "Temp"),
…Automatically Download the Latest Software Release from GitHub with Python
Use the GitHub API to fetch the latest release metadata and download the first asset (binary or archive) to a local directory.
import requests
import sys
from pathlib import Path
def download_latest_release(owner: str, repo: str, output_dir: str = ".") -> None:
"""Download the latest release asset from a GitHub repository."""
url = f"https://api.github.com/repos/{owner}/{repo}/releases/latest"
response = requests.get(url)
res…Benchmark File Read and Write Speed in Python
Measures file write and read throughput in MB/s by writing and reading a temporary file of a given size.
import os
import time
import tempfile
def benchmark_write(file_path, size_mb=100):
data = b'x' * (1024 * 1024) # 1 MB block
start = time.perf_counter()
with open(file_path, 'wb') as f:
for _ in range(size_mb):
f.write(data)
elapsed = time.perf_counter() - start
return size_mb …Build a Complete Web Scraper with Requests and BeautifulSoup in Python
Scrape multiple paginated pages from a website using Requests and BeautifulSoup, with retry logic, error handling, and CSV export.
import requests
from bs4 import BeautifulSoup
import csv
import time
from typing import List, Dict, Optional
class WebScraper:
def __init__(self, base_url: str, output_file: str = "scraped_data.csv"):
self.base_url = base_url
self.output_file = output_file
self.session = requests.Session()…Build a Complete Website Sitemap Generator Without External Services
Crawl a website recursively using only Python's standard library to generate a structured sitemap of internal links.
import json
from urllib.parse import urlparse, urljoin
from collections import deque
import urllib.request
import urllib.error
import re
from html.parser import HTMLParser
class SitemapParser(HTMLParser):
def __init__(self, base_url):
super().__init__()
self.base_url = base_url
self.links …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.