event-driven-architecture

name	Event-Driven Architecture
description	Software architecture pattern using events to trigger communication between decoupled services
category	software-development

Event-Driven Architecture

What I do

I provide a design pattern where system components communicate through the production and consumption of events. EDA enables loose coupling between services, allowing them to evolve independently. Events represent significant occurrences in the system—state changes, domain events, or integration signals. Services react to events asynchronously, enabling scalability, resilience, and real-time processing capabilities.

When to use me

Use EDA when services need to communicate without tight coupling, when real-time processing is required, or when multiple consumers need the same information. It's ideal for microservices architectures, complex event processing, and systems requiring scalability. EDA shines when you have multiple independent components that need to stay synchronized. Avoid it for simple request-response workflows or when synchronous behavior is required.

Core Concepts

• Event: Something that happened in the system
• Event Producer: Service that generates events
• Event Consumer: Service that processes events
• Event Channel: Transport mechanism for events
• Event Broker: Middleware managing event distribution
• Pub/Sub Model: Multiple consumers subscribe to event types
• Event Sourcing: Storing state changes as event sequence
• CQRS: Separating read and write models
• Saga Pattern: Managing distributed transactions
• Idempotency: Processing events safely multiple times
• Event Ordering: Maintaining sequence for related events

Code Examples

Basic Event System

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol, TypeVar, Generic
from uuid import UUID, uuid4
import json

T = TypeVar("T")

@dataclass
class Event:
    event_id: UUID
    event_type: str
    occurred_at: datetime
    payload: dict

class EventPublisher(Protocol):
    def publish(self, event: Event) -> None:
        pass

class EventConsumer(Protocol):
    def handle(self, event: Event) -> None:
        pass

class InMemoryEventBus:
    def __init__(self):
        self._subscribers: dict[str, list[EventConsumer]] = {}
        self._events: list[Event] = []
    
    def subscribe(self, event_type: str, handler: EventConsumer) -> :
         event_type   ._subscribers:
            ._subscribers[event_type] = []
        ._subscribers[event_type].append(handler)
    
     () -> :
        ._events.append(event)
        event_type = event.event_type
         event_type  ._subscribers:
             handler  ._subscribers[event_type]:
                :
                    handler.handle(event)
                 Exception  e:
                    ()

 ():
     ():
        ().__init__(
            event_id=uuid4(),
            event_type=event_type,
            occurred_at=datetime.utcnow(),
            payload=payload
        )

 :
     ():
        ._publisher = publisher
        ._users: [UUID, ] = {}
    
     () -> UUID:
        user_id = uuid4()
        ._users[user_id] = {: user_id, : name, : email}
        
        event = DomainEvent(
            event_type=,
            payload={: (user_id), : email}
        )
        ._publisher.publish(event)
         user_id
    
     () -> :
         user_id   ._users:
             ValueError()
        old_email = ._users[user_id][]
        ._users[user_id][] = new_email
        
        event = DomainEvent(
            event_type=,
            payload={
                : (user_id),
                : old_email,
                : new_email
            }
        )
        ._publisher.publish(event)

 ():
     () -> :
         event.event_type == :
            ()
         event.event_type == :
            ()

Message Broker with RabbitMQ

import pika
import json
from dataclasses import asdict
from datetime import datetime
from typing import Callable
from uuid import uuid4

class RabbitMQEventBus:
    def __init__(
        self,
        host: str = "localhost",
        queue_prefix: str = "events_"
    ):
        self.host = host
        self.queue_prefix = queue_prefix
        self._connection: pika.BlockingConnection | None = None
        self._channel: pika.channel.Channel | None = None
        self._handlers: dict[str, list[Callable]] = {}
    
    def connect(self) -> None:
        credentials = pika.PlainCredentials("guest", "guest")
        parameters = pika.ConnectionParameters(
            host=self.host,
            credentials=credentials
        )
        self._connection = pika.BlockingConnection(parameters)
        self._channel = self._connection.channel()
    
    def publish(self, exchange: , routing_key: , message: ) -> :
         ._connection    ._connection.is_closed:
            .connect()
        
        ._channel.basic_publish(
            exchange=exchange,
            routing_key=routing_key,
            body=json.dumps(message),
            properties=pika.BasicProperties(
                delivery_mode=,
                content_type=,
                timestamp=(datetime.utcnow().timestamp())
            )
        )
    
     () -> :
         ._connection    ._connection.is_closed:
            .connect()
        
        ._channel.exchange_declare(
            exchange=exchange,
            exchange_type=,
            durable=
        )
        
        full_queue = 
        ._channel.queue_declare(queue=full_queue, durable=)
        ._channel.queue_bind(
            exchange=exchange,
            queue=full_queue,
            routing_key=routing_key
        )
        
         ():
            :
                message = json.loads(body)
                handler(message)
                channel.basic_ack(delivery_tag=method.delivery_tag)
             Exception  e:
                ()
                channel.basic_nack(delivery_tag=method.delivery_tag, requeue=)
        
        ._channel.basic_consume(
            queue=full_queue,
            on_message_callback=on_message
        )
    
     () -> :
         ._channel:
            ._channel.start_consuming()
    
     () -> :
         ._connection   ._connection.is_closed:
            ._connection.close()

 :
     ():
        .bus = bus
        .exchange = 
    
     () -> :
        message = {
            : ,
            : order_id,
            : customer_id,
            : total,
            : datetime.utcnow().isoformat()
        }
        .bus.publish(
            exchange=.exchange,
            routing_key=,
            message=message
        )
    
     () -> :
        message = {
            : ,
            : order_id,
            : tracking,
            : datetime.utcnow().isoformat()
        }
        .bus.publish(
            exchange=.exchange,
            routing_key=,
            message=message
        )

Async Processing with Kafka

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
from datetime import datetime
from typing import Optional

class KafkaEventProducer:
    def __init__(self, bootstrap_servers: list[str]):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode("utf-8"),
            key_serializer=lambda k: k.encode("utf-8") if k else None
        )
    
    def send(
        self,
        topic: str,
        key: str,
        value: dict,
        partition: Optional[int] = None
    ) -> None:
        future = self.producer.send(
            topic,
            key=key,
            value=value,
            partition=partition
        )
        try:
            record_metadata = future.get(timeout=10)
            print(f"Sent to {record_metadata.topic}[{record_metadata.partition}]")
        except KafkaError as e:
            print(f"Error sending: ")
    
     () -> :
        event = {
            : event_type,
            : payload,
            : datetime.utcnow().isoformat()
        }
        .send(topic, key=event_type, value=event)
    
     () -> :
        .producer.flush()
        .producer.close()

 :
     ():
        .consumer = KafkaConsumer(
            *topics,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            value_deserializer= v: json.loads(v.decode()),
            auto_offset_reset=,
            enable_auto_commit=
        )
    
     () -> :
         message  .consumer:
            :
                handler(message.topic, message.value)
             Exception  e:
                ()
    
     () -> :
        .consumer.close()

 :
     () -> :
         item  order_data.get(, []):
            ._reserve_inventory(
                product_id=item[],
                quantity=item[],
                order_id=order_data[]
            )
    
     () -> :
        ()

Event Processing Pipeline

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Callable
from uuid import uuid4

@dataclass
class ProcessedEvent:
    event_id: str
    event_type: str
    processed_at: datetime
    result: dict

class EventProcessor(ABC):
    @abstractmethod
    def can_process(self, event_type: str) -> bool:
        pass
    
    @abstractmethod
    def process(self, event: dict) -> ProcessedEvent:
        pass

class EnrichmentProcessor(EventProcessor):
    def __init__(self, enrichment_service):
        self._service = enrichment_service
    
    def can_process(self, event_type: str) -> bool:
        return event_type == "OrderCreated"
    
    def process() -> ProcessedEvent:
        enriched = ._service.enrich_order(event)
         ProcessedEvent(
            event_id=(uuid4()),
            event_type=event[],
            processed_at=datetime.utcnow(),
            result=enriched
        )

 ():
     () -> :
         event_type  [, ]
    
     () -> ProcessedEvent:
        errors = ._validate(event)
         errors:
             ValueError()
         ProcessedEvent(
            event_id=(uuid4()),
            event_type=event[],
            processed_at=datetime.utcnow(),
            result={: }
        )
    
     () -> []:
        errors = []
          event.get():
            errors.append()
          event.get():
            errors.append()
         errors

 :
     ():
        ._processors: [EventProcessor] = []
    
     () -> :
        ._processors.append(processor)
    
     () -> [ProcessedEvent]:
        event_type = event.get(, )
        results = []
        
         processor  ._processors:
             processor.can_process(event_type):
                :
                    result = processor.process(event)
                    results.append(result)
                 Exception  e:
                    ()
        
         results

Saga Implementation

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Protocol
from uuid import uuid4

class SagaStepStatus(Enum):
    PENDING = "pending"
    COMPLETED = "completed"
    FAILED = "failed"
    COMPENSATING = "compensating"

@dataclass
class SagaContext:
    saga_id: str
    data: dict
    completed_steps: list[str] = []

class Saga(ABC):
    def __init__(self):
        self.saga_id = str(uuid4())
        self.context: SagaContext | None = None
    
    @abstractmethod
    def get_steps(self) -> list['SagaStep']:
        pass
    
    def execute(self, initial_data: dict) -> bool:
        self.context = SagaContext(
            saga_id=self.saga_id,
            data=initial_data
        )
        
         step  .get_steps():
            :
                step.execute(.context)
                .context.completed_steps.append(step.name)
             Exception  e:
                ()
                ._compensate(step.name)
                 
        
         
    
     () -> :
        steps = .get_steps()
         step  (steps):
             step.name == failed_step_name:
                
             step.name  .context.completed_steps:
                :
                    step.compensate(.context)
                 Exception  e:
                    ()

 ():


     () -> :
        
    

     () -> :
        
    

     () -> :
        

 ():
     () -> [SagaStep]:
         [
            ValidateOrderStep(),
            ReserveInventoryStep(),
            ProcessPaymentStep(),
            CreateShipmentStep()
        ]

 ():

     () -> :
         
    
     () -> :
        ()
        context.data[] = 
    
     () -> :
        ()

 ():

     () -> :
         
    
     () -> :
        ()
        context.data[] = 
    
     () -> :
        ()
        context.data[] = 

Best Practices

1. Design for Failure: Expect and handle event processing failures
2. Idempotency: Handle duplicate events safely
3. Event Ordering: Use partitioning for ordering guarantees
4. Schema Management: Use event schemas (Avro, Protobuf)
5. Dead Letter Queues: Capture unprocessable events
6. Monitoring: Track event processing latency and errors
7. Testing: Test event consumers in isolation
8. Versioning: Support multiple event schema versions
9. Consumer Groups: Scale consumers horizontally
10. Replayability: Support event replay for debugging
11. CQRS Consideration: Use EDA naturally with CQRS
12. Saga Management: Implement distributed transactions carefully