Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Wrap Message Attributes in a CloudEvent with Python
Create a minimal CloudEvent dataclass that wraps arbitrary message attributes into a JSON envelope, matching CloudEvents 1.0 spec.
import json
from dataclasses import dataclass, field, asdict
from typing import Any, Dict
from datetime import datetime, timezone
@dataclass
class CloudEvent:
message_attributes: Dict[str, Any] = field(default_factory=dict)
def wrap(self, event_id: str, source: str, event_type: str, data: Any):
self…Cache Penetration Null Object Mock in Python
Implement a cache that stores a null marker on misses to prevent repeated database hits, reducing cache penetration.
import time
from collections import defaultdict
from typing import Any, Optional
class Cache:
def __init__(self):
self.store: dict[str, Any] = {}
self.ttl: dict[str, float] = {}
self.null_marker = object()
def get(self, key: str, ttl: int = 60, fallback:
Any = None) -> An…How to Build a Bloom Filter to Reduce Cache Misses in Python
Implement a probabilistic Bloom filter in Python that lets a cache quickly determine which keys are definitely not present, reducing expensive source lookups on cache misses.
import hashlib
import random
class BloomFilter:
def __init__(self, size=100, num_hashes=3):
self.size = size
self.num_hashes = num_hashes
self.bit_array = [0] * size
def _hashes(self, item):
result = []
for i in range(self.num_hashes):
hash_value = int(hash…How to Implement a Negative Cache with TTL in Python
This code provides a TTL mock cache that stores negative results (cache misses) for a short time to reduce repeated lookups of missing keys.
from time import time, sleep
class TTLMockCache:
def __init__(self, ttl_seconds=5):
self.ttl = ttl_seconds
self.store = {}
self.negative_cache = {}
def get(self, key):
now = time()
if key in self.store:
value, expires_at = self.store[key]
if exp…How to Implement an LFU Cache in Python
Implement a Least Frequently Used (LFU) cache with frequency tracking dictionaries to evict the least accessed items when capacity is reached.
class LFUCache:
def __init__(self, capacity: int):
self.capacity = capacity
self.data = {}
self.freq = {}
self.min_freq = 0
def get(self, key: int) -> int:
if key not in self.data:
return -1
self._increment_freq(key)
return self.data[key]
…Implement a Multi-Level Cache with L1 Memory and L2 Redis in Python
This code implements a simple multi-level cache with an in-process L1 cache (via functools.lru_cache) and a mock Redis L2 cache with TTL, falling back to a slow computation on misses.
import time
from functools import lru_cache
class MockRedis:
def __init__(self):
self.store = {}
def get(self, key):
return self.store.get(key, None)
def set(self, key, value, ttl=5):
self.store[key] = (value, time.time() + ttl)
def get_ttl(self, key):
value, expiry…At Least Once with Idempotent Consumer in Python
Implements a thread-safe idempotent consumer that processes each unique message exactly once, even when a producer sends duplicates under an at-least-once delivery model.
import threading
import time
import uuid
from collections import Counter
class IdempotentConsumer:
def __init__(self):
self.processed = set()
self._lock = threading.Lock()
def consume(self, message_id, payload):
with self._lock:
if message_id in self.processed:
…Exactly Once Processing Dedupe Mock in Python
Implements a streaming deduplicator using a set and queue to guarantee each item is processed exactly once while preserving insertion order.
from collections import deque
class DedupeStream:
def __init__(self):
self.seen = set()
self.queue = deque()
def add(self, item):
if item not in self.seen:
self.seen.add(item)
self.queue.append(item)
print(f"Processed: {item} (exactly once)")
…How to Stop Receiving Requests Until Ready in Python
A mock server that refuses requests until a readiness gate is passed, simulating fail-stop behavior for production reliability.
import random
import time
class MockServer:
def __init__(self):
self.ready = False
self.requests_received = 0
def readiness_check(self):
"""Simulates a readiness probe. Returns True only when ready."""
if not self.ready:
return False
return True
def r…How to implement an idempotency key store in Python
Build an in-memory idempotency key store with TTL that processes a request once and reuses the cached result for duplicate calls.
import hashlib
import time
from typing import Dict, Optional
class IdempotencyStore:
"""Simple in-memory idempotency key store with mock processing."""
def __init__(self, ttl_seconds: int = 3600) -> None:
self.ttl = ttl_seconds
self._store: Dict[str, tuple[str, float]] = {}
def _is_expi…Export Metrics with OTLP Mock in Python
Simulates system metric collection and exports them as an OTLP-like JSON payload using only Python's standard library.
from dataclasses import dataclass, asdict
import json
import random
import time
@dataclass
class Metric:
name: str
value: float
timestamp: int
unit: str = "1"
def collect_system_metrics() -> list[Metric]:
"""Mock metric collection for OTLP export simulation."""
now = int(time.time())
re…How to Add Metadata Attributes to a Span in Python
Create a lightweight dataclass-based Span mock that stores key-value metadata attributes for tracing or event logging.
from dataclasses import dataclass, field
from typing import Dict, Any
@dataclass
class Span:
name: str
attributes: Dict[str, Any] = field(default_factory=dict)
def set_attribute(self, key: str, value: Any) -> None:
self.attributes[key] = value
def get_attribute(self, key: str) -> Any…How to Model Span Events in Python
Define a Span class with timestamped milestone events and a completion marker to track operation lifecycle.
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import List
class SpanStatus(Enum):
STARTED = "started"
COMPLETED = "completed"
@dataclass
class SpanEvent:
name: str
timestamp: float = field(default_factory=time.time)
attributes: dict = field(default_facto…CQRS with Separate Read and Write Repositories in Python
Implement CQRS in Python with separate write and read repositories, using commands for mutations and frozen DTOs for queries.
from dataclasses import dataclass
from typing import Dict, List, Optional
# --- Write side: commands mutate state ---
@dataclass
class CreateUserCommand:
id: int
name: str
class UserWriteRepository:
def __init__(self) -> None:
self._store: Dict[int, Dict[str, object]] = {}
def create(self,…Consumer Driven Contract Pact Mock in Python
Define and verify consumer-driven contracts using Pact's Consumer and Provider classes, mocking the provider to assert expected interactions.
from pact import Consumer, Provider
pact = Consumer('OrderService').has_pact_with(Provider('InventoryService'))
@Pact.verify()
class TestInventoryContract:
def test_get_inventory(self):
expected = {"item": "widget", "quantity": 100}
(pact
.given('inventory exists for widget')
.u…How to Implement a Data Helper for Microservices in Python
Create a reusable helper class to serialize, deserialize, and wrap data for microservice communication using dataclasses and JSON.
import json
from dataclasses import dataclass, asdict
from typing import Any, Dict, List
@dataclass
class ServiceResponse:
status: str
data: Any
message: str = ""
class DataHelper:
"""Simple helper for microservice data handling."""
@staticmethod
def serialize(data: Dict[str, Any]) -> str:…How to Mock a Choreography Saga in Python
Simulate a choreography-based saga with event envelopes, status tracking, and compensating actions to model distributed transactions.
import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from enum import Enum
class SagaStatus(Enum):
PENDING = "PENDING"
COMPLETING = "COMPLETING"
COMPLETED = "COMPLETED"
FAILED = "FAILED"
@dataclass
class EventEnvelope:
event_type: str
order_id: str
sta…How to Mock a GraphQL Backend in Python
Create an in-memory GraphQL mock backend using dataclasses and resolver methods returning plain dictionaries.
from dataclasses import dataclass, asdict
from typing import Any, Dict, List
@dataclass
class Product:
id: int
name: str
price: float
@dataclass
class User:
id: int
username: str
class MockGraphQLBackend:
def __init__(self) -> None:
self.products = [
Product(id=1, name…How to implement the Database per service pattern in Python
Simulate separate databases per microservice in Python using dataclasses and in-memory dictionaries, showing how services own their data independently.
import json
from dataclasses import dataclass, asdict
from typing import Dict, List
@dataclass
class User:
id: int
name: str
email: str
@dataclass
class Order:
id: int
user_id: int
product: str
amount: float
class UserServiceDB:
"""Simulates a separate database for the User servic…Idempotent Consumer Event Processing in Python
Track processed event IDs to skip duplicates and count event types for a reliable, idempotent consumer.
import json
from collections import defaultdict
class EventProcessor:
def __init__(self):
self.processed_ids = set()
self.counts = defaultdict(int)
def process_event(self, event):
event_id = event["id"]
if event_id in self.processed_ids:
return {"status": "skipped"…How to Mock Spark Streaming Micro-Batches in Python
Simulate Spark's micro-batch streaming with a simple deque-based class that collects events over time and processes them in timed batches.
import time
from collections import deque
from datetime import datetime
class MicroBatchStream:
def __init__(self, batch_interval_sec=2):
self.batch_interval = batch_interval_sec
self.source = deque()
self.processed = []
def add_events(self, events):
self.source.extend(events…How to Mock a Compute-Collect Action Trigger in Python
Mock a compute-collect action trigger using Python's unittest.mock to simulate Spark-style job execution and assert trigger behavior.
Here's a Python code sample for the problem title "Action trigger compute collect mock":How to Build a Data Validation Schema in Python
Create a lightweight validation schema using dataclasses and lambda validators to check fields in a dictionary.
import re
from dataclasses import dataclass, field
from typing import Any, Callable
@dataclass
class Field:
name: str
validator: Callable[[Any], bool]
required: bool = True
def validate(self, value: Any) -> bool:
if not self.required and value is None:
return True
return …How to Define Dagster ML Assets in Python
Define a chain of Dagster software-defined assets that compute raw features, normalized features, and predictions for an ML pipeline.
from dagster import asset
@asset
def raw_features():
return {"sepal_length": [5.1, 4.9, 6.2], "sepal_width": [3.5, 3.0, 3.4]}
@asset
def normalized_features(raw_features):
values = raw_features["sepal_length"]
mean = sum(values) / len(values)
std = (sum((x - mean) ** 2 for x in values) / len(values…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.