import time from functools import wraps class CircuitBreaker: def __init__(self, failure_threshold=5, recovery_timeout=30, exceptions=(Exception,)): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.exceptions = exceptions self.state = "CLOSED" self.failure_count = 0 self.last_failure_time = None def __call__(self, func): @wraps(func) def wrapper(*args, **kwargs): if self.state == "OPEN": if time.time() - self.last_failure_time > self.recovery_timeout: self.state = "HALF-OPEN" else: raise RuntimeError(f"Service unavailable - circuit breaker open") try: result = func(*args, **kwargs) # Reset on successful HALF-OPEN call  if self.state == "HALF-OPEN": self.failure_count = 0 self.state = "CLOSED" return result except self.exceptions as e: self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = "OPEN" raise e return wrapper # Usage example @CircuitBreaker(failure_threshold=3, recovery_timeout=60) def call_external_api(endpoint): # Make API call that might fail  response = requests.get(endpoint) response.raise_for_status() return response.json() 

import redis import time import uuid class DistributedLock: def __init__(self, redis_client, lock_name, expire_seconds=10): self.redis = redis_client self.lock_name = lock_name self.expire_seconds = expire_seconds self.identifier = str(uuid.uuid4()) def __enter__(self): self.acquired = self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb): if self.acquired: self.release() def acquire(self): # Use SET NX to acquire the lock atomically  acquired = self.redis.set( self.lock_name, self.identifier, nx=True, ex=self.expire_seconds ) # Start lock refresh thread if acquired  if acquired: self._start_refresh_thread() return acquired def release(self): # Release the lock only if we own it (using Lua script for atomicity)  script = """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end """ self.redis.eval(script, 1, self.lock_name, self.identifier) if hasattr(self, 'refresh_thread'): self.refresh_thread_stop = True def _start_refresh_thread(self): import threading self.refresh_thread_stop = False def refresh_lock(): while not self.refresh_thread_stop: # Refresh lock if we still own it  script = """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end """ self.redis.eval( script, 1, self.lock_name, self.identifier, self.expire_seconds ) time.sleep(self.expire_seconds / 3) self.refresh_thread = threading.Thread(target=refresh_lock) self.refresh_thread.daemon = True self.refresh_thread.start() # Usage example redis_client = redis.Redis(host='localhost', port=6379, db=0) def process_account(account_id): lock_name = f"account_lock:{account_id}" with DistributedLock(redis_client, lock_name, expire_seconds=30) as lock: if not lock.acquired: print(f"Account {account_id} is being processed by another worker") return # Critical section - only one process will execute this at a time  print(f"Processing account {account_id}") time.sleep(10) # Simulate work  print(f"Finished processing account {account_id}") 

import pika import json import time import logging class MessageProcessor: def __init__(self, amqp_url, queue_name, dead_letter_queue, max_retries=3, retry_delay=5): self.amqp_url = amqp_url self.queue_name = queue_name self.dead_letter_queue = dead_letter_queue self.max_retries = max_retries self.retry_delay = retry_delay self.logger = logging.getLogger(__name__) def setup_queues(self): connection = pika.BlockingConnection( pika.URLParameters(self.amqp_url) ) channel = connection.channel() # Declare the dead-letter queue  channel.queue_declare(queue=self.dead_letter_queue, durable=True) # Declare the main queue with dead-letter exchange  channel.queue_declare( queue=self.queue_name, durable=True, arguments={ 'x-dead-letter-exchange': '', 'x-dead-letter-routing-key': self.dead_letter_queue } ) connection.close() def publish_message(self, message): connection = pika.BlockingConnection( pika.URLParameters(self.amqp_url) ) channel = connection.channel() channel.basic_publish( exchange='', routing_key=self.queue_name, body=json.dumps(message), properties=pika.BasicProperties( delivery_mode=2, # make message persistent  headers={'retry_count': 0} ) ) connection.close() def process_messages(self, callback): connection = pika.BlockingConnection( pika.URLParameters(self.amqp_url) ) channel = connection.channel() def wrapped_callback(ch, method, properties, body): try: # Extract retry count  headers = properties.headers or {} retry_count = headers.get('retry_count', 0) # Process the message  message = json.loads(body) self.logger.info(f"Processing message: {message}") callback(message) # Acknowledge successful processing  ch.basic_ack(delivery_tag=method.delivery_tag) except Exception as e: self.logger.error(f"Error processing message: {e}") # Increment retry count  retry_count = retry_count + 1 if retry_count <= self.max_retries: # Reject message for requeue with updated retry count  time.sleep(self.retry_delay) # Republish with incremented retry count  ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_publish( exchange='', routing_key=self.queue_name, body=body, properties=pika.BasicProperties( delivery_mode=2, headers={'retry_count': retry_count} ) ) else: # Max retries reached, send to dead-letter queue  self.logger.warning( f"Max retries reached, sending to dead-letter queue" ) ch.basic_ack(delivery_tag=method.delivery_tag) # Add failure reason to the message  message = json.loads(body) message['_failure_reason'] = str(e) channel.basic_publish( exchange='', routing_key=self.dead_letter_queue, body=json.dumps(message), properties=pika.BasicProperties(delivery_mode=2) ) # Consume messages one at a time  channel.basic_qos(prefetch_count=1) channel.basic_consume( queue=self.queue_name, on_message_callback=wrapped_callback ) self.logger.info(f"Waiting for messages on {self.queue_name}") channel.start_consuming() def process_dead_letters(self, callback): connection = pika.BlockingConnection( pika.URLParameters(self.amqp_url) ) channel = connection.channel() def wrapped_callback(ch, method, properties, body): try: message = json.loads(body) self.logger.info(f"Processing dead letter: {message}") callback(message) ch.basic_ack(delivery_tag=method.delivery_tag) except Exception as e: self.logger.error(f"Error processing dead letter: {e}") # Nack without requeue to keep in dead-letter queue  ch.basic_nack( delivery_tag=method.delivery_tag, requeue=False ) channel.basic_qos(prefetch_count=1) channel.basic_consume( queue=self.dead_letter_queue, on_message_callback=wrapped_callback ) self.logger.info(f"Waiting for messages on {self.dead_letter_queue}") channel.start_consuming() 

import threading import time import random import requests from enum import Enum class NodeState(Enum): FOLLOWER = 1 CANDIDATE = 2 LEADER = 3 class RaftNode: def __init__(self, node_id, cluster_nodes, election_timeout_range=(150, 300)): self.node_id = node_id self.cluster_nodes = cluster_nodes self.election_timeout_range = election_timeout_range # Persistent state  self.current_term = 0 self.voted_for = None self.log = [] # Volatile state  self.state = NodeState.FOLLOWER self.commit_index = 0 self.last_applied = 0 # Leader state  self.next_index = {node: 1 for node in cluster_nodes if node != node_id} self.match_index = {node: 0 for node in cluster_nodes if node != node_id} # Initialize timers  self.election_timer = None self.heartbeat_timer = None self.reset_election_timer() # Start the main thread  self.running = True threading.Thread(target=self.apply_entries_thread).start() def reset_election_timer(self): if self.election_timer: self.election_timer.cancel() timeout = random.randint( self.election_timeout_range[0], self.election_timeout_range[1] ) self.election_timer = threading.Timer( timeout / 1000.0, self.start_election ) self.election_timer.daemon = True self.election_timer.start() def start_heartbeat(self): if self.heartbeat_timer: self.heartbeat_timer.cancel() # Send heartbeats immediately  self.send_append_entries() # Schedule next heartbeat  self.heartbeat_timer = threading.Timer( 50 / 1000.0, # 50ms  self.start_heartbeat ) self.heartbeat_timer.daemon = True self.heartbeat_timer.start() def start_election(self): if not self.running: return self.state = NodeState.CANDIDATE self.current_term += 1 self.voted_for = self.node_id votes_received = 1 # Vote for self  print(f"Node {self.node_id} starting election for term {self.current_term}") # Reset election timer  self.reset_election_timer() # Request votes from all other nodes  for node in self.cluster_nodes: if node == self.node_id: continue try: last_log_index = len(self.log) last_log_term = self.log[-1]['term'] if self.log else 0 response = requests.post( f"http://{node}/request_vote", json={ 'term': self.current_term, 'candidate_id': self.node_id, 'last_log_index': last_log_index, 'last_log_term': last_log_term }, timeout=0.1 ) result = response.json() # If response contains higher term, revert to follower  if result['term'] > self.current_term: self.current_term = result['term'] self.state = NodeState.FOLLOWER self.voted_for = None return if result['vote_granted']: votes_received += 1 # Check if majority achieved  if votes_received > len(self.cluster_nodes) / 2: self.become_leader() return except Exception as e: print(f"Error requesting vote from {node}: {e}") def become_leader(self): if not self.running or self.state == NodeState.LEADER: return print(f"Node {self.node_id} becoming leader for term {self.current_term}") self.state = NodeState.LEADER # Initialize leader state  self.next_index = { node: len(self.log) + 1 for node in self.cluster_nodes if node != self.node_id } self.match_index = { node: 0 for node in self.cluster_nodes if node != self.node_id } # Cancel election timer  if self.election_timer: self.election_timer.cancel() # Start sending heartbeats  self.start_heartbeat() def send_append_entries(self): if not self.running or self.state != NodeState.LEADER: return for node in self.cluster_nodes: if node == self.node_id: continue try: next_idx = self.next_index[node] prev_log_index = next_idx - 1 prev_log_term = ( self.log[prev_log_index - 1]['term'] if prev_log_index > 0 and self.log else 0 ) # Get entries to send  entries = self.log[prev_log_index:] if prev_log_index < len(self.log) else [] response = requests.post( f"http://{node}/append_entries", json={ 'term': self.current_term, 'leader_id': self.node_id, 'prev_log_index': prev_log_index, 'prev_log_term': prev_log_term, 'entries': entries, 'leader_commit': self.commit_index }, timeout=0.1 ) result = response.json() # If response contains higher term, revert to follower  if result['term'] > self.current_term: self.current_term = result['term'] self.state = NodeState.FOLLOWER self.voted_for = None # Cancel heartbeat timer  if self.heartbeat_timer: self.heartbeat_timer.cancel() # Reset election timer  self.reset_election_timer() return if result['success']: # Update indices  if entries: self.match_index[node] = prev_log_index + len(entries) self.next_index[node] = self.match_index[node] + 1 # Update commit index if needed  self.update_commit_index() else: # Decrement next_index and retry  self.next_index[node] = max(1, self.next_index[node] - 1) except Exception as e: print(f"Error sending append entries to {node}: {e}") def update_commit_index(self): if self.state != NodeState.LEADER: return # Find the highest index that has been replicated to a majority of nodes  for n in range(self.commit_index + 1, len(self.log) + 1): # Count nodes that have this entry  count = 1 # Leader has it  for node in self.match_index: if self.match_index[node] >= n: count += 1 # Check if majority reached and entry is from current term  if count > len(self.cluster_nodes) / 2 and ( n > 0 and self.log[n - 1]['term'] == self.current_term): self.commit_index = n def append_log(self, data): if self.state != NodeState.LEADER: return False, "Not the leader" # Append to local log  entry = { 'term': self.current_term, 'data': data } self.log.append(entry) # Update leader's match index for itself  self.match_index[self.node_id] = len(self.log) # Try to replicate immediately  self.send_append_entries() return True, "Log entry added" def apply_entries_thread(self): while self.running: # Apply committed entries that haven't been applied yet  while self.last_applied < self.commit_index: self.last_applied += 1 entry = self.log[self.last_applied - 1] # Apply the command (in a real system, this would modify state)  print(f"Node {self.node_id} applying: {entry['data']}") time.sleep(0.01) # Small sleep to prevent CPU hogging  def request_vote_handler(self, term, candidate_id, last_log_index, last_log_term): if term < self.current_term: return { 'term': self.current_term, 'vote_granted': False } if term > self.current_term: self.current_term = term self.state = NodeState.FOLLOWER self.voted_for = None # Check if log is at least as up-to-date as ours  our_last_index = len(self.log) our_last_term = self.log[-1]['term'] if self.log else 0 log_ok = ( last_log_term > our_last_term or (last_log_term == our_last_term and last_log_index >= our_last_index) ) if (self.voted_for is None or self.voted_for == candidate_id) and log_ok: # Grant vote  self.voted_for = candidate_id self.reset_election_timer() return { 'term': self.current_term, 'vote_granted': True } else: return { 'term': self.current_term, 'vote_granted': False } def append_entries_handler(self, term, leader_id, prev_log_index, prev_log_term, entries, leader_commit): # Reply false if term < currentTerm  if term < self.current_term: return { 'term': self.current_term, 'success': False } # If term is greater, update current term and convert to follower  if term > self.current_term: self.current_term = term self.state = NodeState.FOLLOWER self.voted_for = None # Reset election timer since we've heard from the leader  self.reset_election_timer() # If candidate or leader, step down  if self.state != NodeState.FOLLOWER: self.state = NodeState.FOLLOWER if self.heartbeat_timer: self.heartbeat_timer.cancel() # Check if our log has the prev_log_entry with matching term  if prev_log_index > 0: if prev_log_index > len(self.log) or ( prev_log_index > 0 and self.log[prev_log_index - 1]['term'] != prev_log_term): return { 'term': self.current_term, 'success': False } # Process entries  if entries: # Find conflicts: if an existing entry conflicts with a new one,  # delete it and all that follow  for i, entry in enumerate(entries): log_index = prev_log_index + i + 1 # If we have an entry at this index but terms don't match  if log_index <= len(self.log) and self.log[log_index - 1]['term'] != entry['term']: # Delete this and all subsequent entries  self.log = self.log[:log_index - 1] break # Append any new entries not already in the log  for i, entry in enumerate(entries): log_index = prev_log_index + i + 1 if log_index > len(self.log): self.log.append(entry) # Update commit index  if leader_commit > self.commit_index: self.commit_index = min(leader_commit, len(self.log)) return { 'term': self.current_term, 'success': True } def shutdown(self): self.running = False if self.election_timer: self.election_timer.cancel() if self.heartbeat_timer: self.heartbeat_timer.cancel() 

 python import uuid import json import datetime from collections import defaultdict class EventStore: def __init__(self, persistence_adapter=None): self.persistence_adapter = persistence_adapter self.events = defaultdict(list) self.event_handlers = defaultdict(list) def save_event(self, aggregate_id, event_type, event_data): """Save an event to the store""" event = { 'id': str(uuid.uuid4()), 'timestamp': datetime.datetime.now().isoformat(), 'aggregate_id': aggregate_id, 'type': event_type, 'data': event_data, 'sequence': len(self.events[aggregate_id]) } # Add to in-memory store self.events[aggregate_id].append(event) # Persist if adapter available if self.persistence_adapter: self.persistence_adapter.save_event(event) # Trigger event handlers for handler in self.event_handlers[event_type]: handler(event) return event def get_events(self, aggregate_id): """Get all events for an aggregate""" if self.persistence_adapter: # Load from persistence if adapter available return self.persistence_adapter.get_events(aggregate_id) return self.events[aggregate_id] def register_handler(self, event_type, handler): """Register a handler for an event type""" self.event_handlers[event_type].append(handler) class Aggregate: def __init__(self, aggregate_id, event_store): self.id = aggregate_id self.event_store = event_store self.version = -1 # Apply existing events to rebuild state self.replay_events() def replay_events(self): """Replay all events to rebuild aggregate state""" events = self.event_store.get_events(self.id) for event in events: self.apply_event(event, replay=True) self.version = event['sequence'] def apply_event(self, event, replay=False): """Apply an event to the aggregate""" event_type = event['type'] handler_method = f"apply_{event_type}" if hasattr(self, handler_method): getattr(self, handler_method)(event['data'], replay) def create_event(self, event_type, event_data): """Create and save a new event""" event = self.event_store.save_event( self.id, event_type, event_data ) # Apply the event to update current state self.apply_event(event) self.version = event['sequence'] # Example: SQLite persistence adapter class SQLiteEventStore: def __init__(self, db_path): import sqlite3 self.conn = sqlite3.connect(db_path) self.create_tables() def create_tables(self): cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS events ( id TEXT PRIMARY KEY, timestamp TEXT, aggregate_id TEXT, type TEXT, data TEXT, sequence INTEGER, UNIQUE(aggregate_id, sequence) ) ''') self.conn.commit() def save_event(self, event): cursor = self.conn.cursor() cursor.execute( ''' INSERT INTO events (id, timestamp, aggregate_id, type, data, sequence) VALUES (?, ?, ?, ?, ?, ?) ''', ( event['id'], event['timestamp'], event['aggregate_id'], event['type'], json.dumps(event['data']), event['sequence'] ) ) self.conn.commit() def get_events(self, aggregate_id): cursor = self.conn.cursor() cursor.execute( ''' SELECT id, timestamp, aggregate_id, type, data, sequence FROM events WHERE aggregate_id = ? ORDER BY sequence ''', (aggregate_id,) ) events = [] for row in cursor.fetchall(): events.append({ 'id': row[0], 'timestamp': row[1], 'aggregate_id': row[2], 'type': row[3], 'data': json.loads(row[4]), 'sequence': row[5] }) return events # Example: Bank Account Aggregate class BankAccount(Aggregate): def __init__(self, account_id, event_store): self.balance = 0 self.status = "new" super().__init__(account_id, event_store) def open_account(self, customer_id, initial_deposit): if self.status != "new": raise ValueError("Account already opened") if initial_deposit < 100: raise ValueError("Initial deposit must be at least $100") self.create_event("account_opened", { "customer_id": customer_id, "initial_deposit": --- ## 101 Books **101 Books** is an AI-driven publishing company co-founded by author **Aarav Joshi**. By leveraging advanced AI technology, we keep our publishing costs incredibly low—some books are priced as low as **$4**—making quality knowledge accessible to everyone. Check out our book **[Golang Clean Code](https://www.amazon.com/dp/B0DQQF9K3Z)** available on Amazon. Stay tuned for updates and exciting news. When shopping for books, search for **Aarav Joshi** to find more of our titles. Use the provided link to enjoy **special discounts**! ## Our Creations Be sure to check out our creations: **[Investor Central](https://www.investorcentral.co.uk/)** | **[Investor Central Spanish](https://spanish.investorcentral.co.uk/)** | **[Investor Central German](https://german.investorcentral.co.uk/)** | **[Smart Living](https://smartliving.investorcentral.co.uk/)** | **[Epochs & Echoes](https://epochsandechoes.com/)** | **[Puzzling Mysteries](https://www.puzzlingmysteries.com/)** | **[Hindutva](http://hindutva.epochsandechoes.com/)** | **[Elite Dev](https://elitedev.in/)** | **[JS Schools](https://jsschools.com/)** --- ### We are on Medium **[Tech Koala Insights](https://techkoalainsights.com/)** | **[Epochs & Echoes World](https://world.epochsandechoes.com/)** | **[Investor Central Medium](https://medium.investorcentral.co.uk/)** | **[Puzzling Mysteries Medium](https://medium.com/puzzling-mysteries)** | **[Science & Epochs Medium](https://science.epochsandechoes.com/)** | **[Modern Hindutva](https://modernhindutva.substack.com/)**