Design a Notification System
You'll learn to
- -Use Observer so notification producers never need to know which channels (email, SMS, push) are subscribed
- -Add per-channel formatting via Strategy and retry/failure handling without coupling it to the core notify path
A notification system is close to a pure application of the Observer pattern covered earlier in this course - working through it end to end is mostly about layering in the realistic details (per-channel formatting, retries, user preferences) around that Observer core without letting them leak into each other.
Step 1: The Observer Core
class NotificationEvent:
def __init__(self, event_type: str, user_id: str, data: dict):
self.event_type = event_type
self.user_id = user_id
self.data = data
class NotificationChannel(ABC):
@abstractmethod
def send(self, event: NotificationEvent) -> None: ...
class NotificationService:
def __init__(self):
self._channels: list[NotificationChannel] = []
def subscribe(self, channel: NotificationChannel) -> None:
self._channels.append(channel)
def publish(self, event: NotificationEvent) -> None:
for channel in self._channels:
channel.send(event) # publisher has zero knowledge of channel internalsStep 2: Per-Channel Formatting as Strategy
An order-shipped event needs to render very differently as an email (subject line, HTML body) versus an SMS (a single short line of text) versus a push notification (a title and truncated body). Each NotificationChannel owns its own MessageFormatter, keeping formatting logic local to the channel that needs it rather than centralizing every format's logic in one sprawling class.
class MessageFormatter(ABC):
@abstractmethod
def format(self, event: NotificationEvent) -> str: ...
class SMSFormatter(MessageFormatter):
def format(self, event: NotificationEvent) -> str:
return f"Update: {event.data.get('summary', event.event_type)}"
class EmailChannel(NotificationChannel):
def __init__(self, formatter: MessageFormatter):
self._formatter = formatter
def send(self, event: NotificationEvent) -> None:
body = self._formatter.format(event)
deliver_email(event.user_id, subject=event.event_type, body=body)Step 3: Retry Without Coupling It to publish()
A channel's delivery can fail (an email provider times out) without that failure blocking or corrupting delivery to other channels - each channel.send() call should be independently fault-isolated, typically wrapped by the NotificationService in a try/except that logs the failure and continues to the next channel, and each channel can independently implement its own retry-with-backoff internally without NotificationService needing to know anything about retry policy at all.
class NotificationService:
def publish(self, event: NotificationEvent) -> None:
for channel in self._channels:
try:
channel.send(event)
except DeliveryError as e:
log_failure(channel, event, e) # isolated failure, loop continues
continueStep 4: User Preferences (Opt-Out Per Channel)
A realistic follow-up: users can opt out of SMS but keep email. Rather than branching inside publish(), the cleaner extension wraps subscription itself: a PreferenceFilteredChannel decorator (recall Decorator from earlier) wraps any NotificationChannel and checks the user's preferences before delegating send() to the real channel - preferences become one more layer, not a special case scattered through the core loop.
Naming the fault-isolation detail explicitly - "one channel's failure should never prevent delivery through the others" - is exactly the kind of production-mindedness detail that separates a strong answer from a merely correct one on this prompt.
Interview Signal is part of Pro
See a real weak answer next to a real strong one for this exact topic.
Quiz is part of Pro
Test what you just read with a short quiz, and bank the XP.