Reference library

Microservices patterns

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

4 matches

Bulkhead Thread Pool per Service Mock in Python

Simulates a bulkhead pattern with per-service thread pools and semaphore-based rejection to isolate failures between dependent services.

bulkhead threadpool semaphore

Python

import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor

class ServiceBulkhead:
    def __init__(self, name, max_threads, max_queue):
        self.name = name
        self.executor = ThreadPoolExecutor(max_workers=max_threads)
        self.semaphore = threading.Semaphore(max_thread…

13 0 Open

Microservices patterns medium

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.

cqrs repositories microservices

Python

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,…

16 0 Open

Microservices patterns medium

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.

microservices database-per-service dataclasses

Python

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…

12 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.