🐍 RabbitMQ with Python

SOLID Principles, Clean Architecture & Best Practices

📋 Project Overview

🏗️

SOLID Principles

Every component follows Single Responsibility, Open/Closed, Liskov Substitution, Interface Segregation, and Dependency Inversion principles

🏛️

Clean Architecture

Clear separation of concerns with Core, Infrastructure, Services, and Models layers

🔄

Event-Driven

Domain events with proper validation, routing, and error handling

💉

Dependency Injection

Proper DI container with service registry for testability and maintainability

📊

Monitoring

Health checks, metrics collection, and performance monitoring built-in

🛡️

Error Handling

Comprehensive exception handling with retry logic and resilience patterns

📈 Implementation Stats

15+
Classes
8
Interfaces
3
Services
6
Event Types
100%
Type Safe

🏗️ SOLID Principles Implementation

S

Single Responsibility

Each class has one reason to change

class UserService: # Only handles user domain logic
O

Open/Closed

Open for extension, closed for modification

class IEventHandler(ABC): # Easy to add new handlers
L

Liskov Substitution

Subtypes must be substitutable

publisher: IMessagePublisher # Any implementation works
I

Interface Segregation

Many specific interfaces vs one general

IMessagePublisher # Publishing IMessageConsumer # Consuming
D

Dependency Inversion

Depend on abstractions, not concretions

def __init__(self, publisher: IEventPublisher): self.publisher = publisher

🎬 SOLID in Action

IEventPublisher
→
EventPublisher
→
ProducerService

🏛️ Clean Architecture Layers

🎯 Core

Interfaces
Exceptions
Domain Logic

📊 Models

Event DTOs
Validation
Factories

⚙️ Services

User Service
Order Service
Payment Service

🔧 Infrastructure

RabbitMQ
Monitoring
Registry

📜 Dependency Rules

✅ Infrastructure → Services → Core
❌ Core ← Services ← Infrastructure
✅ All layers can use Models
✅ Only Infrastructure knows RabbitMQ

⚙️ Design Patterns

🏭 Factory Pattern

EventFactory creates validated events

EventFactory.create_event(EventType.USER_CREATED, **data)

👁️ Observer Pattern

Event handlers observe and react to domain events

registry.register_handler("user.created", handler)

📝 Registry Pattern

Service registry manages dependencies

registry.get_service("user_service")

🎯 Strategy Pattern

Different exchange types use different routing strategies

DirectExchange | FanoutExchange | TopicExchange

📋 Template Method

BaseEvent defines common structure

class UserCreatedEvent(BaseEvent):

🔌 Adapter Pattern

RabbitMQ adapter implements messaging interfaces

class RabbitMQPublisher(IMessagePublisher):

🔄 Message Flow Animation

📤
Producer
↓
Event
↓
🔄
Publisher
↓
Routing Key
↓
📋
RabbitMQ
↓
Distributing...
↙
↓
↘
👤
User
📦
Order
💳
Payment

📝 Flow Details

💻 Code Examples

🔌 Core Interfaces

class IEventPublisher(ABC):
    """Interface for publishing domain events"""

    @abstractmethod
    async def publish_event(
        self,
        event_type: str,
        event_data: Dict[str, Any],
        correlation_id: Optional[str] = None
    ) -> None:
        pass

class IEventHandler(ABC):
    """Interface for handling domain events"""

    @abstractmethod
    async def handle(
        self,
        event_data: Dict[str, Any],
        metadata: MessageMetadata
    ) -> None:
        pass

    @abstractmethod
    def can_handle(self, routing_key: str) -> bool:
        pass

⚙️ Service Implementation

class UserEventHandler(IEventHandler):
    """Handles user-related events following SRP"""

    def can_handle(self, routing_key: str) -> bool:
        return routing_key in ["user.created", "user.updated"]

    async def handle(self, event_data: Dict[str, Any], metadata: MessageMetadata):
        event_type = event_data.get("event_type")

        if event_type == "user.created":
            await self._handle_user_created(event_data, metadata)
        elif event_type == "user.updated":
            await self._handle_user_updated(event_data, metadata)

    async def _handle_user_created(self, event_data, metadata):
        # Business logic for user creation
        event = UserCreatedEvent(**event_data)
        await self._create_user_profile(event)
        await self._send_welcome_email(event)

📊 Event Models

class UserCreatedEvent(BaseEvent):
    """User creation event with validation"""

    event_type: str = Field(default="user.created", const=True)
    user_id: str
    email: str
    first_name: str
    last_name: str
    registration_source: Optional[str] = "web"

    @validator("email")
    def validate_email(cls, v):
        if "@" not in v:
            raise ValueError("Invalid email format")
        return v.lower()

class EventFactory:
    """Factory for creating validated events"""

    @classmethod
    def create_event(cls, event_type: EventType, **kwargs) -> BaseEvent:
        event_class = cls._event_mapping.get(event_type)
        if not event_class:
            raise ValueError(f"Unknown event type: {event_type}")
        return event_class(**kwargs)

🎯 Usage Examples

# Publishing events
await producer_service.user_registered(
    user_id="user123",
    email="john@example.com",
    first_name="John",
    last_name="Doe",
    correlation_id="req-456"
)

# Service setup with DI
connection_manager = RabbitMQConnectionManager(settings)
publisher = RabbitMQPublisher(connection_manager)
event_publisher = EventPublisher(publisher)
producer_service = ProducerService(event_publisher)

# Register handlers
registry.register_handler("user.created", user_handler)
registry.register_handler("order.placed", order_handler)

# Health monitoring
health = await health_monitor.check_system_health()
print(f"System status: {health.status}")