Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
Observer Pattern in Python: Notify Listeners
Implement the Observer design pattern in Python with a Subject class that manages listeners and notifies them with messages.
class Subject:
def __init__(self):
self._observers = []
def attach(self, observer):
self._observers.append(observer)
def detach(self, observer):
self._observers.remove(observer)
def notify(self, message):
for observer in self._observers:
observer.update(me…Build a Live Countdown Timer for Events in Python
A Python script that displays a real-time countdown to a target date and time, updating every second in the console.
import datetime
import time
def countdown(event_name, target_datetime):
"""Displays a live countdown to a target datetime."""
while True:
now = datetime.datetime.now()
remaining = target_datetime - now
if remaining.total_seconds() <= 0:
print(f"\n🚀 {event_name} is happening…Generate a Monthly Report CSV from Log Files in Python
Reads a CSV log file, filters events by a given month, aggregates daily event counts and revenue, and writes a summarized monthly report to a new CSV.
import csv
from collections import defaultdict
from datetime import datetime
def generate_monthly_report(log_file: str, month: str, output_file: str) -> None:
events_by_date = defaultdict(int)
revenue_by_date = defaultdict(float)
with open(log_file, 'r') as f:
for line in f:
date_…Track Internet Connectivity and Downtime Automatically in Python
Monitors internet connectivity by pinging a remote host and logs any downtime events with timestamps and duration.
import time
import subprocess
from datetime import datetime
def check_internet(host="8.8.8.8", timeout=3):
"""Returns True if internet is reachable via ping."""
try:
subprocess.run(
["ping", "-c", "1", "-W", str(timeout), host],
capture_output=True,
timeout=timeout …Deduplicate events by ID within a window in Python
Deduplicate event streams by ID within sliding time windows, keeping the newest occurrence per window using heaps and sets.
import heapq
from collections import defaultdict
def deduplicate_events(events, window_size):
"""Return events deduplicated by id, keeping newest within each sliding window."""
# Index events by (timestamp, id) for deterministic ordering
events_by_id = defaultdict(list)
for ts, eid, *payload in events…Enrich Events with Geo IP Data in Python
Returns a copy of each event dictionary, enriched with a geo-location dict from a mock IP-to-geo lookup table, with a fallback for unknown IPs.
import ipaddress
GEO_IP_DB = {
"192.168.1.10": {"country": "US", "city": "New York", "lat": 40.7128, "lon": -74.0060},
"10.0.0.5": {"country": "DE", "city": "Berlin", "lat": 52.5200, "lon": 13.4050},
"172.16.0.8": {"country": "JP", "city": "Tokyo", "lat": 35.6762, "lon": 139.6503},
}
EVENTS = [
{"id…Group Python Events into Sessions with a Gap Timeout
Groups timestamped events into sessions, starting a new session when the time gap exceeds a specified timeout.
from itertools import groupby
from datetime import datetime, timedelta
def session_window_group(events, gap_seconds=300):
"""Group events into sessions where gap > gap_seconds starts a new session."""
if not events:
return []
events = sorted(events, key=lambda x: x[0])
sessions = []
c…How to Count Events by Minute with a Tumbling Window in Python
Group timestamps into fixed 60-second tumbling windows and count events per bucket using a dict.
from collections import defaultdict
from datetime import datetime, timedelta
def tumbling_window_count(events, window_seconds=60):
buckets = defaultdict(int)
for event in events:
ts = datetime.fromisoformat(event["timestamp"])
bucket_start = ts - timedelta(seconds=ts.second % window_seconds,
…How to Deduplicate Events with At-Least-Once Delivery in Python
Implements an exactly-once processing pattern for at-least-once event delivery by tracking seen event IDs in a set, skipping duplicates.
seen_ids = set()
def process_event(event_id: str, payload: dict) -> dict:
"""Process an event exactly once, ignoring duplicates."""
if event_id in seen_ids:
return {"status": "duplicate", "event_id": event_id}
seen_ids.add(event_id)
return {"status": "processed", "event_id": event_id, **payloa…How to route late-arriving data to a side output in Python
Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.
from collections import defaultdict
def late_arriving_side_output(events, late_threshold_ts):
"""
Mock a streaming pipeline that separates late-arriving data events
into a side output list (e.g., for dead-letter analysis).
events: list of (timestamp, data) tuples, timestamps as ints.
late_thresho…Generate Mock CloudFormation Stack Events in Python
Generate a list of mock AWS CloudFormation stack events with random resources, statuses, and timestamps, and print them as JSON.
import json
import random
from datetime import datetime, timedelta
def generate_mock_stack_events(stack_name="MyTestStack", num_events=10):
"""Generate a list of mock CloudFormation stack events."""
resources = [
("AWS::S3::Bucket", "MyBucket"),
("AWS::EC2::Instance", "MyInstance"),
("…How to Mock Auto Scaling Policy Scale Out in Python
Define a mock auto-scaling function that scales out capacity by a factor up to a max, simulating AWS-like events.
def mock_scale_out(current_capacity: int, max_capacity: int, scale_factor: int = 1) -> tuple:
"""
Mock auto-scaling policy: scales out by the specified factor
if capacity allows, capped at max_capacity.
"""
if current_capacity >= max_capacity:
return current_capacity, False
new_cap…How to Mock GCP Cloud Functions HTTP Events in Python
Simulate a GCP Cloud Functions HTTP event with a Python mock handler that constructs a realistic event payload and returns a JSON response.
import json
from datetime import datetime, timezone
def mock_http_event(data):
"""Simulate a GCP Cloud Function HTTP event."""
event = {
"event_id": "mock-event-12345",
"timestamp": datetime.now(timezone.utc).isoformat(),
"event_type": "google.cloud.functions.http",
"resource"…How to Initialize Sentry SDK with a Mock DSN in Python
Initialize the Sentry SDK in Python with a mock DSN to test error tracking without sending real events, then verify the DSN configuration.
import sentry_sdk
# Initialize Sentry SDK with a mock DSN (no real events will be sent)
sentry_sdk.init(
dsn="https://mock-public@mock-host/mock-project",
traces_sample_rate=1.0,
environment="development",
)
# Capture a test message to confirm SDK is configured
sentry_sdk.capture_message("Test message fr…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 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 Build an Append-Only Event Store in Python
Implement a simple append-only event store class that stores events in a list and supports retrieval by index range.
class EventStore:
def __init__(self):
self._events = []
def append(self, event):
"""Append an event to the store."""
self._events.append(event)
def get_events(self, start=0, end=None):
"""Return events from start index to end (exclusive)."""
return self._events[sta…Outbox pattern reliable publish in Python with SQLite
Implements a transactional outbox with SQLite, ensuring reliable message publishing by storing events in the same DB transaction as business changes.
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
class Outbox:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTO…How to mock Server-Sent Events (SSE) in Python
A minimal HTTP server that streams Server-Sent Events to clients, perfect for testing and development.
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time
MESSAGES = iter([
"data: Hello world\n\n",
"data: Second message\n\n",
"event: custom\n",
"data: Custom event payload\n\n",
"data: Final message\n\n"
])
class SSEHandler(BaseHTTPRequestHandler):
def do_GET…Event sourcing append store replay in Python
A simple in-memory event store that appends events per aggregate and replays them on demand.
import json
from collections import defaultdict
class EventStore:
def __init__(self):
self._events = defaultdict(list)
def append(self, aggregate_id, event_type, data):
event = {"type": event_type, "data": data}
self._events[aggregate_id].append(event)
def replay(self, aggregate…How to Build a Materialized View Updater Consumer Mock in Python
A mock consumer that queues change events and triggers refresh callbacks to simulate materialized view updates.
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional
@dataclass
class MaterializedViewUpdater:
"""Mock updater that consumes change events and refreshes a view."""
refresh: Optional[Callable[[str], None]] = None
queue: Deque[tuple…How to Build a Mock Change Data Capture Event Stream in Python
Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.
from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta
seed(42) # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)
def mock_cdc_event(stream_index: int) -> dict:
…How to Implement Publish-Subscribe Fanout with Multiple Subscribers in Python
Create a simple publish-subscribe system in Python that broadcasts messages to multiple subscriber callbacks for a given topic.
import time
class PubSub:
def __init__(self):
self.subscribers = {}
def subscribe(self, topic, callback):
if topic not in self.subscribers:
self.subscribers[topic] = []
self.subscribers[topic].append(callback)
def publish(self, topic, message):
if topic in sel…How to Implement a Tumbling Window Counter in Python
Count events that fall within a fixed-size sliding time window using a deque and pruning logic.
from collections import deque
import time
class TumblingWindowCounter:
def __init__(self, window_size_seconds):
self.window_size = window_size_seconds
self.window = deque()
def add_event(self, timestamp):
self.window.append(timestamp)
def count(self, current_time):
while…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.