Observer pattern

Quality Score

Overall Score: 8.6/10 ✅ Excellent

• Technical Accuracy: 30/35
• Code Quality: 24/25
• Educational Value: 20/25
• Documentation: 12/15

Last reviewed: June 26, 2026

The observer pattern is a behavioral design pattern used when you want to notify multiple objects about a change of state/event in one object. The relationship is one-to-many, so when the state of the subject changes, all its observers are notified and updated automatically. The benefits of this pattern are:

• Decoupling: The subject and observers are loosely coupled, allowing for independent development and maintenance.
• Dynamic relationships: Observers can be added or removed at runtime, allowing for flexible and dynamic relationships between objects.
• Broadcast communication: The subject can notify multiple observers simultaneously, enabling efficient communication and reducing the need for direct dependencies between objects.
• Event-driven architecture: The observer pattern is commonly used in event-driven systems, where changes in one component trigger updates in other components.
• Main use case: When you need to notify different objects about changes in the state of another object.
• Avoid using the pattern when the relationship is one-to-one.

The observer pattern is composed of three main components:

Subject

Maintains a list of observers and provides methods to attach, detach, and notify them. It is responsible for managing the state and notifying observers of any changes.

src.advanced.observer_pattern.Subject

Subject that maintains observers and notifies them of changes.

Source code in src/advanced/observer_pattern/observer.py

class Subject:
    """Subject that maintains observers and notifies them of changes."""

    def __init__(self) -> None:
        """Initialize the subject with an empty list of observers."""
        self._observers: list[Observer] = []

    def subscribe(self, observer: Observer) -> None:
        """Add an observer to the subject.

        Args:
            observer (Observer): The observer to subscribe.
        """
        self._observers.append(observer)
        logger.info("Subscribed an observer: %s", observer.__class__.__name__)

    def unsubscribe(self, observer: Observer) -> None:
        """Remove an observer from the subject.

        Args:
            observer (Observer): The observer to unsubscribe.
        """
        self._observers.remove(observer)
        logger.info("Unsubscribed an observer: %s", observer.__class__.__name__)

    def notify(self, message: str) -> None:
        """Notify all subscribed observers with a message.

        Args:
            message (str): The message to send to all observers.
        """
        logger.info("Notifying observers...")
        for observer in self._observers:
            observer.update(message)

Observer Interface

Defines the update method that observers must implement to receive notifications from the subject. It establishes a contract for communication between the subject and observers.

src.advanced.observer_pattern.Observer

Bases: ABC

Abstract base class for observers.

Source code in src/advanced/observer_pattern/observer.py

class Observer(ABC):
    """Abstract base class for observers."""

    @abstractmethod
    def update(self, message: str) -> None:
        """Receive an update from the subject.

        Args:
            message (str): The notification message from the subject.
        """
        pass

Concrete Observers

Implement the observer interface and define the specific behavior to be executed when notified of a state change in the subject. Each concrete observer can have its own unique response to the notification.

src.advanced.observer_pattern.KafkaObserver

Bases: Observer

Concrete observer that simulates sending messages to Kafka.

Source code in src/advanced/observer_pattern/observer.py

class KafkaObserver(Observer):
    """Concrete observer that simulates sending messages to Kafka."""

    def __init__(self, settings: dict) -> None:
        """Init the KafkaObserver with specific settings.

        Args:
            settings (dict): Configuration settings for the Kafka client.
        """
        self.settings = settings
        self._client = None  # Simulate a Kafka client

    def update(self, message: str) -> None:
        """Simulate sending a message to Kafka.

        Args:
            message (str): The message to send to Kafka.
        """
        logger.info("KafkaObserver received message: %s", message)

src.advanced.observer_pattern.RabbitMQObserver

Bases: Observer

Concrete observer that simulates sending messages to RabbitMQ.

Source code in src/advanced/observer_pattern/observer.py

class RabbitMQObserver(Observer):
    """Concrete observer that simulates sending messages to RabbitMQ."""

    def __init__(self, settings: dict) -> None:
        """Init the RabbitMQObserver with specific settings.

        Args:
            settings (dict): Configuration settings for the RabbitMQ client.
        """
        self.settings = settings
        self._client = None  # Simulate a RabbitMQ client

    def update(self, message: str) -> None:
        """Simulate sending a message to RabbitMQ.

        Args:
            message (str): The message to send to RabbitMQ.
        """
        logger.info("RabbitMQObserver received message: %s", message)

Some real-world examples are: event-driven systems, notification systems, pub-sub architectures, real-time data updates...