Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to implement saga orchestration with compensating steps in Python
Orchestrate a distributed transaction across services, rolling back completed steps with compensations when a later step fails.
class InventoryService:
def reserve(self, order_id):
print(f"[Inventory] Reserving stock for order {order_id}")
return True
def compensate(self, order_id):
print(f"[Inventory] Releasing stock for order {order_id}")
class PaymentService:
def charge(self, order_id):
print(f…Implement the Transactional Outbox Pattern with SQLite in Python
A Python implementation of the transactional outbox pattern using SQLite, ensuring atomic writes of order data and outbox events in a single transaction while supporting reliable message publishing and consumption.
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
import json
@dataclass
class Order:
order_id: str
amount: float
status: str
class TransactionalOutbox:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self._create_tab…How to Mock a Redis Transaction with MULTI/EXEC in Python
A minimal in-memory mock of Redis MULTI/EXEC transactions that queues commands and applies them atomically on EXEC.
class RedisTransactionMock:
def __init__(self):
self.data = {}
self.queue = []
self.in_transaction = False
def multi(self):
self.in_transaction = True
self.queue = []
return "OK"
def set(self, key, value):
if self.in_transaction:
self.qu…Python Redis WATCH optimistic lock mock
A MockRedis class that simulates WATCH/MULTI/EXEC transactions with optimistic locking to detect concurrent modifications before committing.
import time
import threading
class MockRedis:
def __init__(self):
self.data = {}
self.watched = {}
self.lock = threading.Lock()
def get(self, key):
return self.data.get(key)
def set(self, key, value):
self.data[key] = value
def watch(self, *keys):
wi…Mock a Two-Phase Commit Coordinator in Python
Simulates a two-phase commit protocol where a coordinator asks participants to prepare, then commits or aborts based on unanimous readiness.
import random
import time
from typing import Dict, List
class TwoPhaseCommitCoordinator:
def __init__(self, participants: List[str]):
self.participants = participants
self.participant_state: Dict[str, bool] = {}
def prepare(self) -> bool:
print("[Coordinator] Phase 1: Prepare")
…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…Python Saga Compensating Steps Mock
Mock a distributed transaction saga with forward steps and compensating actions that reverse partial progress on failure.
from datetime import datetime
def make_payment(user_id, amount):
print(f"[{datetime.now():%H:%M:%S}] Payment of ${amount} processed for user {user_id}")
return {"step": "payment", "status": "ok", "details": f"${amount} charged"}
def deduct_inventory(order_id, items):
print(f"[{datetime.now():%H:%M:%S}]…How to Mock a Cross-Shard Saga in Python
Simulate a distributed saga with compensating transactions across multiple database shards using a lightweight Python class that tracks executed steps and rolls them back in reverse on failure.
import json
class SagaState:
def __init__(self, saga_id):
self.saga_id = saga_id
self.executed_steps = []
self.compensations = []
def execute_step(self, shard, step_name, operation):
self.executed_steps.append((shard, step_name))
print(f"[Saga {self.saga_id}] Executin…How to Simulate Distributed Transactions in Python with a Mock
Model distributed transaction behavior with a mock Transaction class that supports commit, rollback, and failure simulation.
class Transaction:
def __init__(self, id):
self.id = id
self.operations = []
self.committed = False
def add_operation(self, op, data):
self.operations.append((op, data))
def commit(self):
if not self.operations:
raise ValueError("No operations to commit…How to mock batch commit of transactions in Python
Simulate a transaction batch writer with commit, rollback, and summary logic to test database write patterns without a real database.
import json
from datetime import datetime, timezone
class TransactionBatch:
def __init__(self):
self.pending = []
self.committed = []
self._log = []
def add(self, operation):
self.pending.append(operation)
def commit(self):
if not self.pending:
return …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.