System Design Expert Expert guidance for system design, software architecture, scalability patterns, and distributed systems. Core Concepts Architecture Patterns Microservices vs Monolithic Event-driven architecture CQRS and Event Sourcing Layered architecture Hexagonal architecture Service-oriented architecture (SOA) Scalability Horizontal vs vertical scaling Load balancing strategies Caching layers Database sharding Read replicas CDN usage Distributed Systems CAP theorem Consistency models Distributed consensus (Raft, Paxos) Message queues Service discovery Circuit breakers Design Patterns

Singleton Pattern

class DatabaseConnection : _instance = None _lock = threading . Lock ( ) def new ( cls ) : if cls . _instance is None : with cls . _lock : if cls . _instance is None : cls . _instance = super ( ) . new ( cls ) cls . _instance . _initialize ( ) return cls . _instance def _initialize ( self ) : self . connection = self . _create_connection ( )

Factory Pattern

class ShapeFactory : @staticmethod def create_shape ( shape_type : str ) : if shape_type == "circle" : return Circle ( ) elif shape_type == "square" : return Square ( ) raise ValueError ( f"Unknown shape: { shape_type } " )

Observer Pattern

class Subject : def init ( self ) : self . _observers = [ ] def attach ( self , observer ) : self . _observers . append ( observer ) def notify ( self , event ) : for observer in self . _observers : observer . update ( event )

Strategy Pattern

class PaymentStrategy : def pay ( self , amount ) : pass class CreditCardPayment ( PaymentStrategy ) : def pay ( self , amount ) : return f"Paid $ { amount } via credit card" class PayPalPayment ( PaymentStrategy ) : def pay ( self , amount ) : return f"Paid $ { amount } via PayPal" Scalability Patterns

Circuit Breaker Pattern

from enum import Enum import time class CircuitState ( Enum ) : CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" class CircuitBreaker : def init ( self , failure_threshold = 5 , timeout = 60 ) : self . failure_threshold = failure_threshold self . timeout = timeout self . failure_count = 0 self . last_failure_time = None self . state = CircuitState . CLOSED def call ( self , func , * args , ** kwargs ) : if self . state == CircuitState . OPEN : if time . time ( ) - self . last_failure_time

self . timeout : self . state = CircuitState . HALF_OPEN else : raise Exception ( "Circuit breaker is OPEN" ) try : result = func ( * args , ** kwargs ) self . on_success ( ) return result except Exception as e : self . on_failure ( ) raise e def on_success ( self ) : self . failure_count = 0 self . state = CircuitState . CLOSED def on_failure ( self ) : self . failure_count += 1 self . last_failure_time = time . time ( ) if self . failure_count = self . failure_threshold : self . state = CircuitState . OPEN

Rate Limiter

from collections import deque import time class RateLimiter : def init ( self , max_requests , window_seconds ) : self . max_requests = max_requests self . window_seconds = window_seconds self . requests = deque ( ) def allow_request ( self , user_id ) : now = time . time ( )

Remove old requests outside window

while self . requests and self . requests [ 0 ] [ 1 ] < now - self . window_seconds : self . requests . popleft ( )

Check if under limit

user_requests

sum ( 1 for uid , _ in self . requests if uid == user_id ) if user_requests < self . max_requests : self . requests . append ( ( user_id , now ) ) return True return False Caching Strategy from functools import wraps import hashlib import json class CacheStrategy : """Implement caching patterns""" def init ( self , cache_backend ) : self . cache = cache_backend def cache_aside ( self , key , fetch_func , ttl = 3600 ) : """Cache-aside (lazy loading)""" data = self . cache . get ( key ) if data is None : data = fetch_func ( ) self . cache . set ( key , data , ttl ) return data def write_through ( self , key , data , persist_func ) : """Write-through caching""" self . cache . set ( key , data ) persist_func ( data ) def write_behind ( self , key , data , queue ) : """Write-behind (write-back) caching""" self . cache . set ( key , data ) queue . enqueue ( lambda : self . persist ( key , data ) ) def memoize ( ttl = 3600 ) : """Memoization decorator""" cache = { } def decorator ( func ) : @wraps ( func ) def wrapper ( * args , ** kwargs ) : key = hashlib . md5 ( json . dumps ( ( args , kwargs ) , sort_keys = True ) . encode ( ) ) . hexdigest ( ) if key in cache : cached_value , timestamp = cache [ key ] if time . time ( ) - timestamp < ttl : return cached_value result = func ( * args , ** kwargs ) cache [ key ] = ( result , time . time ( ) ) return result return wrapper return decorator Database Patterns

Database Sharding

class ShardRouter : def init ( self , num_shards ) : self . num_shards = num_shards self . shards = [ f"shard_ { i } " for i in range ( num_shards ) ] def get_shard ( self , key ) : """Route to shard based on key""" shard_id = hash ( key ) % self . num_shards return self . shards [ shard_id ]

Read Replica Pattern

class DatabaseRouter : def init ( self , primary , replicas ) : self . primary = primary self . replicas = replicas self . current_replica = 0 def execute_write ( self , query ) : """All writes go to primary""" return self . primary . execute ( query ) def execute_read ( self , query ) : """Reads from replicas (round-robin)""" replica = self . replicas [ self . current_replica ] self . current_replica = ( self . current_replica + 1 ) % len ( self . replicas ) return replica . execute ( query ) Load Balancing from typing import List import random class LoadBalancer : """Implement load balancing algorithms""" def init ( self , servers : List [ str ] ) : self . servers = servers self . current = 0 def round_robin ( self ) : """Round-robin load balancing""" server = self . servers [ self . current ] self . current = ( self . current + 1 ) % len ( self . servers ) return server def least_connections ( self , connections_per_server ) : """Least connections algorithm""" return min ( connections_per_server . items ( ) , key = lambda x : x [ 1 ] ) [ 0 ] def random_selection ( self ) : """Random server selection""" return random . choice ( self . servers ) def weighted_round_robin ( self , weights ) : """Weighted round-robin""" total_weight = sum ( weights . values ( ) ) r = random . randint ( 1 , total_weight ) cumulative = 0 for server , weight in weights . items ( ) : cumulative += weight if r <= cumulative : return server Best Practices Design Principles SOLID principles DRY (Don't Repeat Yourself) KISS (Keep It Simple, Stupid) YAGNI (You Aren't Gonna Need It) Separation of concerns Fail fast Design for failure Scalability Plan for growth early Use horizontal scaling Implement caching strategically Async where possible Database optimization Monitor everything Load test regularly Architecture Start with monolith, split when needed Define clear boundaries Use APIs for communication Version APIs properly Document architecture decisions Review regularly Keep it simple Anti-Patterns ❌ Premature optimization ❌ Over-engineering ❌ No monitoring ❌ Tight coupling ❌ God objects/classes ❌ No error handling ❌ Ignoring security Resources System Design Primer: https://github.com/donnemartin/system-design-primer Martin Fowler's Architecture: https://martinfowler.com/architecture/ AWS Architecture: https://aws.amazon.com/architecture/ Microservices.io: https://microservices.io/