Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 36 additions & 11 deletions patterns/behavioral/observer.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,34 +9,59 @@
Flask Signals: https://flask.palletsprojects.com/en/1.1.x/signals/
"""

from __future__ import annotations

from contextlib import suppress
from typing import Protocol
# observer.py

from __future__ import annotations
from typing import List

# define a generic observer type
class Observer(Protocol):
class Observer:
def update(self, subject: Subject) -> None:
"""
Receive update from the subject.

Args:
subject (Subject): The subject instance sending the update.
"""
pass


class Subject:
_observers: List[Observer]

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

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

Args:
observer (Observer): The observer instance to attach.
"""
if observer not in self._observers:
self._observers.append(observer)

def detach(self, observer: Observer) -> None:
with suppress(ValueError):
"""
Detach an observer from the subject.

Args:
observer (Observer): The observer instance to detach.
"""
try:
self._observers.remove(observer)
except ValueError:
pass

def notify(self, modifier: Observer | None = None) -> None:
def notify(self) -> None:
"""
Notify all attached observers by calling their update method.
"""
for observer in self._observers:
if modifier != observer:
observer.update(self)
observer.update(self)


class Data(Subject):
Expand Down