Reference library

Microservices patterns

Service boundaries, discovery, inter-service calls, and decomposition patterns.

5 matches

Distributed tracing with contextvars in Python

Propagate trace and span IDs across function calls using contextvars to mock distributed tracing in a single process.

import contextvars
import uuid
import time

_trace_context = contextvars.ContextVar("trace_context", default=None)


class TraceContext:
    def __init__(self, trace_id, parent_span_id):
        self.trace_id = trace_id
        self.parent_span_id = parent_span_id
        self.span_id = uuid.uuid4().hex[:16]
        s…

13 0 Open

Microservices patterns medium

How to Implement a Two-Phase Commit Mock in Python

Simulate a distributed two-phase commit with prepare, commit, and abort phases, including deterministic failure injection for testing.

2pc transaction microservices

Python

import random
from dataclasses import dataclass
from typing import Dict, List, Optional


@dataclass
class Transaction:
    tx_id: int
    data: Dict[str, str]


class TwoPhaseCommitMock:
    """Simple two-phase commit mock with prepare and commit phases."""

    def __init__(self) -> None:
        self.prepared: List…

13 0 Open

Microservices patterns medium

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.

saga microservices events

Python

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…

13 0 Open

Microservices patterns medium

Python Saga Compensating Steps Mock

Mock a distributed transaction saga with forward steps and compensating actions that reverse partial progress on failure.

saga microservices compensation

Python

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}]…

14 0 Open

Microservices patterns medium

Saga pattern orchestration with rollback in Python

Orchestrate a distributed transaction with Saga steps and automated compensation rollback on failure.

saga microservices transaction

Python

import time
import random


class SagaStep:
    def __init__(self, name):
        self.name = name
        self.executed = False

    def execute(self):
        print(f"Executing {self.name}...")
        time.sleep(0.2)
        if random.random() < 0.3:
            raise RuntimeError(f"{self.name} failed")
        sel…

14 0 Open

Browse by section

Each section groups closely related Python snippets.