Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
Idempotent Consumer: Store Processed IDs in Python
Implement an idempotent consumer that persists processed message IDs to a JSON file, skipping duplicates on restart.
import json
from pathlib import Path
class IdempotentStore:
def __init__(self, storage_path: str = "processed_ids.json"):
self.storage_path = Path(storage_path)
self.processed_ids = self._load()
def _load(self) -> set:
if self.storage_path.exists():
with self.storage_path…Create a Data Helper in Python for gRPC-style APIs
This code builds a simple DataHelper class that mimics gRPC request/response handling with in-memory storage, JSON serialization, and basic CRUD operations for beginners.
import json
from dataclasses import dataclass, asdict
from typing import Dict, Any
@dataclass
class User:
user_id: int
name: str
email: str
class DataHelper:
"""Simple helper to demonstrate gRPC-like data handling for beginners."""
def __init__(self) -> None:
self._users: Dict[int, Use…Format data in Python using dataclasses like gRPC messages
Convert Python dataclasses to and from dicts and format them gRPC-style for clean data handling.
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
@dataclass
class ProductInfo:
"""Data class representing a gRPC-style product message."""
name: str
price: float
tags: List[str]
description: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""C…How to Build a Data Helper Class in Python for Beginners
Create a beginner-friendly DataHelper class that stores, retrieves, filters, and summarizes records in a list of dictionaries.
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
@dataclass
class DataHelper:
"""A beginner-friendly helper for common data tasks."""
data: List[Dict[str, Any]] = field(default_factory=list)
def add_record(self, record…How to Build a Simple Filter Helper in Python for API Design
Create a reusable data filter service with dataclasses that mimics gRPC request/response patterns for filtering dataset records.
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
@dataclass
class FilterRequest:
"""A simple filter request mirroring a gRPC message structure."""
field_name: str
operator: str # eq, ne, gt, lt, contains
value: Any
page_size: int = 10
page_token: Optional…How to Build a Simple gRPC-Style Data Service in Python
Create a beginner-friendly gRPC-style service with dataclasses to simulate GetUser and CreateUser RPCs.
from dataclasses import dataclass
from typing import Optional
@dataclass
class User:
id: int
name: str
email: str
class UserService:
"""Simple gRPC-style service contract for beginner learners."""
def get_user(self, user_id: int) -> Optional[User]:
"""Simulated gRPC GetUser RPC."""
…How to Implement Sparse Fieldsets in Python
A function that filters API responses by resource type, returning only requested fields plus IDs, as a sparse fieldset mock.
from dataclasses import dataclass, field
from typing import Dict, List, Optional
@dataclass
class MockResponse:
data: Dict[str, object] = field(default_factory=dict)
included: List[Dict[str, object]] = field(default_factory=list)
def select_fields(
data: Dict[str, object],
sparse_fields: Optional[D…How to Validate JWT Claims (exp, iss, aud) in Python
This code demonstrates how to decode and validate a JWT's essential claims—expiration (exp), issuer (iss), and audience (aud)—using the PyJWT library, returning clear error messages for common validation failures.
import jwt
from datetime import datetime, timezone, timedelta
SECRET = "mock-secret"
def validate_token(token, expected_iss, expected_aud):
try:
decoded = jwt.decode(
token,
SECRET,
algorithms=["HS256"],
options={"require": ["exp", "iss", "aud"]},
…Sort Python list by query param order_by
Sort a list of dataclass objects dynamically by a field name passed as a query param, with asc/desc direction support.
from dataclasses import dataclass
@dataclass
class Item:
name: str
price: int
def sort_items(items, order_by, direction="asc"):
if order_by not in ("name", "price"):
raise ValueError(f"Unsupported sort field: {order_by}")
reverse = direction.lower() == "desc"
return sorted(items, key=l…Dedupe processed message IDs in Python
Filters an inbox of messages by removing items whose IDs have already been processed, using a set for fast lookups.
from pathlib import Path
import json
def dedupe_processed_ids(inbox_file: Path, processed_file: Path) -> list:
processed = set(json.loads(processed_file.read_text()))
inbox = json.loads(inbox_file.read_text())
deduped = [item for item in inbox if item["id"] not in processed]
return deduped
if __nam…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 Implement a Priority Queue for Messages in Python
Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.
import heapq
from dataclasses import dataclass, field
from typing import Any
@dataclass(order=True)
class Message:
priority: int
sequence: int = field(compare=False)
content: str = field(compare=False)
class PriorityQueue:
def __init__(self):
self._heap = []
def push(self, priority: int,…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…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…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…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 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…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 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…How to Replicate Data Across All Shards in Python
Mocks a global table that replicates a key-value pair to every shard, ensuring reads return the same value from any shard.
from dataclasses import dataclass
from typing import Dict, List
@dataclass
class Shard:
id: str
data: Dict[str, int]
class GlobalTable:
def __init__(self, shards: List[Shard]):
self._shards = {s.id: s for s in shards}
def set_value(self, key: str, value: int) -> None:
"""Replicate …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.