Build a Notification Routing Engine

Design an extensible notification engine using OOP principles in Python.

Introduction

30 Second Summary

Think about how your phone decides which notifications deserve a buzz, which stay silent, and which trigger a full-screen alarm. Behind every smart notification system is a routing engine that matches messages to the right output channel based on priority rules.

In this project, you will build a notification routing engine in Python that dispatches messages to different channels based on priority levels. You'll use object-oriented programming principles like abstract classes, inheritance, and composition to create a system that's easy to extend without modifying existing code.

What You'll Build

A CLI-driven notification routing engine that accepts messages with four priority levels (low, medium, high, critical) and routes them to Console, File, and Webhook channels using polymorphism and composition.

By the end of this project, you'll have:

  • A Notification dataclass and Priority enum that model messages cleanly with automatic timestamps and validation.
  • An abstract base class defining the interface all notification channels must follow, with three concrete implementations.
  • A NotificationRouter that uses composition to manage channels and dispatch notifications polymorphically, proving the Open/Closed Principle in action.
  • Secret Mission: Add a FilterChain that lets channels accept or reject notifications based on custom filter rules like keyword matching and time-of-day restrictions.

Are there any prerequisites?

You should be comfortable with basic Python syntax including variables, functions, loops, and dictionaries. No prior experience with object-oriented programming is required.

Set Up the Project Structure

A well-organized file structure is the foundation of any Python project you want to reuse or extend. Separating your code into a package with dedicated modules keeps each responsibility isolated and makes the whole system easier to test.

In this step, you'll create a Python package called notification_engine with separate module files for models, channels, and routing logic. You'll also confirm your Python version inside Cursor's terminal.

In this step, get ready to:
  • Create the notification-routing-engine project folder and notification_engine package directory.
  • Verify Python 3.12+ is available in Cursor's integrated terminal.
  • Add package-level imports to notification_engine/__init__.py.
Create the project folder and package directory

Your project needs a top-level folder to hold everything, plus a nested notification_engine/ directory that Python recognizes as an importable package.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, navigate into it, and create the package directory by running these commands:
mkdir notification-routing-engine
cd notification-routing-engine
mkdir notification_engine
  • Create all the placeholder files for your package and entry point by running this command:

macOS

touch notification_engine/__init__.py notification_engine/models.py notification_engine/channels.py notification_engine/router.py main.py

Windows

New-Item -ItemType "File" -Path "notification_engine/__init__.py", "notification_engine/models.py", "notification_engine/channels.py", "notification_engine/router.py", "main.py"
  • Confirm the files exist by running:
ls notification_engine

You should see __init__.py, channels.py, models.py, and router.py in the output.

What does this file structure look like?

You now have a project folder with this layout:

  • notification-routing-engine/ is your project root.
  • notification_engine/ is the Python package (the __init__.py file is what tells Python this folder is importable).
  • main.py sits outside the package. It will be your CLI entry point.

Don't see your files?

  • Make sure you ran cd notification-routing-engine before creating the files. If you're in the wrong directory, the files landed elsewhere.
  • Check that you ran mkdir notification_engine before the file-creation command. The directory must exist first.

Help me debug my project directory structure.

Open the project in Cursor and verify Python
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the notification-routing-engine folder, then click Open.

You should see the notification_engine folder and main.py in the Explorer sidebar on the left.

  • Open Cursor's integrated terminal by pressing Ctrl+` (backtick key, same shortcut on macOS and Windows).
  • Verify your Python version by running this command:
python --version

You should see Python 3.12.x or higher (e.g. Python 3.13.2).

Seeing a version lower than 3.12?

On macOS, the system python or python3 may point to an older version. Try running python3 --version instead.

If you need to install or upgrade, visit the official Python downloads page at python.org/downloads and install version 3.12 or higher.

Help me get the right Python version working.

Add the package-level imports to __init__.py

A Python package's __init__.py file controls what's available when someone writes from notification_engine import .... You'll add all the public imports now so the interface is defined upfront.

  • In the Explorer sidebar, expand the notification_engine folder and click __init__.py to open it.
  • Define the package's public API by pasting the following code:
from notification_engine.models import Notification, Priority
from notification_engine.channels import (
    NotificationChannel,
    ConsoleChannel,
    FileChannel,
    WebhookChannel,
)
from notification_engine.router import NotificationRouter

__all__ = [
    "Notification",
    "Priority",
    "NotificationChannel",
    "ConsoleChannel",
    "FileChannel",
    "WebhookChannel",
    "NotificationRouter",
]

What does this code do?

  • The from ... import lines pull specific classes from each module file so users can import them directly from the package (e.g. from notification_engine import Notification).
  • The __all__ list explicitly declares which names are part of the public API. Anything not listed here is considered internal.
  • Save __init__.py.

You won't be able to test the import yet because the classes referenced here don't exist in the module files yet. The next step defines them.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

from notification_engine.models import Notification, Priority
from notification_engine.channels import (
    NotificationChannel,
    ConsoleChannel,
    FileChannel,
    WebhookChannel,
)
from notification_engine.router import NotificationRouter

__all__ = [
    "Notification",
    "Priority",
    "NotificationChannel",
    "ConsoleChannel",
    "FileChannel",
    "WebhookChannel",
    "NotificationRouter",
]

Why define imports before the code exists?

This is an interface-first design approach. By writing the package's public API before the implementation, you define exactly what the outside world will interact with.

Once the module files are populated in the next steps, running python -c "import notification_engine" will confirm everything connects properly.

Your project structure is ready. Next, you'll define the data model that represents notifications with priorities and timestamps.

Define the Data Model with Dataclasses and Enums

Your project skeleton is in place with all the module files and import structure ready to go. Now it's time to fill those empty files with actual Python code.

Every notification system needs a clear definition of what a "notification" is and how urgent it is. In this step, you'll create the data model that every other part of the engine depends on.

In this step, get ready to:
  • Define a Priority enum with four severity levels.
  • Build a Notification dataclass to represent messages.
  • Test the model to confirm everything works.
Define the Priority enum

The router needs a way to compare notification urgency. Python's Enum class lets you define a fixed set of named constants, but it does not support comparison operators like >= or < by default. You'll add those manually so priorities can be compared later.

  • Open notification_engine/models.py in the Cursor editor (it's currently empty).
  • Add the following imports and Priority enum class:
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum


# Define priority levels as an enum with integer values for comparison
class Priority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

    # Custom comparison operators since Enum doesn't support them by default
    def __ge__(self, other):
        if self.__class__ is other.__class__:
            return self.value >= other.value
        return NotImplemented

    def __gt__(self, other):
        if self.__class__ is other.__class__:
            return self.value > other.value
        return NotImplemented

    def __le__(self, other):
        if self.__class__ is other.__class__:
            return self.value <= other.value
        return NotImplemented

    def __lt__(self, other):
        if self.__class__ is other.__class__:
            return self.value < other.value
        return NotImplemented

What does this code do?

The Priority enum defines four fixed severity levels: LOW (1), MEDIUM (2), HIGH (3), and CRITICAL (4). Each level has an integer value so you can compare them numerically.

The four dunder methods (__ge__, __gt__, __le__, __lt__) enable expressions like Priority.HIGH >= Priority.LOW. Without these, Python would raise a TypeError when you try to compare enum members.

The NotImplemented return tells Python to skip the comparison if the other value isn't the same enum type. This prevents accidental comparisons between a Priority and an unrelated object.

Why not use IntEnum?

Python offers IntEnum which automatically supports comparisons. The trade-off is that IntEnum members compare equal to plain integers (Priority.LOW == 1 would be True), which can cause subtle bugs.

Writing the comparison operators manually keeps priorities type-safe. Only Priority members can be compared with each other.

  • Save models.py.
  • You won't see a change yet. The next chunk adds the Notification class that uses this enum.
Define the Notification dataclass

A notification carries a message, a priority level, an optional target channel, and a timestamp. Python's @dataclass decorator auto-generates the __init__ and __repr__ methods for you based on the fields you declare.

  • In models.py, add the following Notification class below the Priority enum:
@dataclass
class Notification:
    message: str
    priority: Priority
    channel_target: str | None = None  # Optional target channel name
    timestamp: datetime = field(default_factory=datetime.now)  # Auto-set to current time

    def __post_init__(self):
        # Validate that the message is not empty or whitespace-only
        if not self.message.strip():
            raise ValueError("Notification message cannot be empty.")

What does this code do?

  • message and priority are required fields. Every notification must have both.
  • channel_target uses str | None union syntax to mark it as optional. It defaults to None if not provided.
  • timestamp uses field(default_factory=datetime.now) to auto-generate the current time when a notification is created.
  • __post_init__ runs automatically after __init__. It validates that the message is not empty or whitespace-only, raising a ValueError if it is.
  • Save models.py.

Seeing a red underline on str | None?

The str | None union syntax requires Python 3.10 or higher. If your editor flags this, confirm you're running 3.12+ by checking python --version in the terminal.

help me fix a type syntax error in my Python models file

✔️ Awesome, I've got everything!

Great. Double check you've saved models.py before moving on.

ⓧ I'd like to double check the full code

Here's the complete notification_engine/models.py file:

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum


class Priority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

    def __ge__(self, other):
        if self.__class__ is other.__class__:
            return self.value >= other.value
        return NotImplemented

    def __gt__(self, other):
        if self.__class__ is other.__class__:
            return self.value > other.value
        return NotImplemented

    def __le__(self, other):
        if self.__class__ is other.__class__:
            return self.value <= other.value
        return NotImplemented

    def __lt__(self, other):
        if self.__class__ is other.__class__:
            return self.value < other.value
        return NotImplemented


@dataclass
class Notification:
    message: str
    priority: Priority
    channel_target: str | None = None
    timestamp: datetime = field(default_factory=datetime.now)

    def __post_init__(self):
        if not self.message.strip():
            raise ValueError("Notification message cannot be empty.")
Test the model

The full package import won't work yet because channels.py and router.py are still empty (__init__.py tries to import from them). You can test models.py directly by adding its directory to the import path.

  • Verify the model works by running this command in the integrated terminal:
python -c "import sys; sys.path.insert(0, 'notification_engine'); from models import Notification, Priority; n = Notification(message='Test', priority=Priority.HIGH); print(n)"
  • You should see output like Notification(message='Test', priority=<Priority.HIGH: 3>, channel_target=None, timestamp=datetime.datetime(...)) with the current date and time auto-populated.

What does this command do?

sys.path.insert(0, 'notification_engine') temporarily adds the package directory to Python's search path. This lets you import models directly without triggering __init__.py (which would fail on the empty modules).

Once you implement channels.py and router.py in later steps, the full package import will work.

  • Now verify the validation logic by running this command:
python -c "import sys; sys.path.insert(0, 'notification_engine'); from models import Notification, Priority; Notification(message='  ', priority=Priority.LOW)"
  • You should see a ValueError: Notification message cannot be empty. traceback. This confirms your __post_init__ validation is working.

Seeing ModuleNotFoundError?

Make sure you're running the command from the project root folder (the one containing the notification_engine/ directory). If you're inside notification_engine/, run cd .. first.

Also confirm that models.py is saved and located at notification_engine/models.py (not in the project root).

help me fix a ModuleNotFoundError when testing my notification models

Your data model is locked in. Next up, you'll build the abstract channel interface that defines how notifications get delivered.

Build the Abstract Channel Interface and Concrete Implementations

Your data model is locked in. You have a Notification that carries a message, a priority level, and a timestamp. Now you need somewhere to send those notifications.

In this step, you will define the contract that every notification channel must follow, then build three concrete channels that honour that contract. The key insight is that the router you build later will never know which specific channel it is talking to. It only knows the interface.

In this step, get ready to:
  • Define a NotificationChannel abstract base class with a send/name contract.
  • Implement three concrete channels: Console, File, and Webhook.
  • Test each channel independently to verify they work.
Define the abstract channel interface

An Abstract Base Class is a blueprint that says "every class inheriting from me must implement these methods." If a subclass forgets one, Python raises an error at instantiation time. This guarantees every channel speaks the same language.

  • Open notification_engine/channels.py in Cursor and add the following code:
from abc import ABC, abstractmethod

from notification_engine.models import Notification, Priority


class NotificationChannel(ABC):
    def __init__(self, min_priority: Priority = Priority.LOW):
        self._min_priority = min_priority

    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @abstractmethod
    def send(self, notification: Notification) -> bool:
        pass

    def supports_priority(self, notification: Notification) -> bool:
        return notification.priority >= self._min_priority

What does this code do?

NotificationChannel inherits from ABC, making it an abstract class. You cannot instantiate it directly.

  • @abstractmethod on send() and name forces every subclass to implement both. Forget one, and Python refuses to create the object.
  • supports_priority() is a concrete method. It checks whether the notification's priority meets the channel's minimum threshold using the comparison operators you built into Priority.
  • self._min_priority uses a leading underscore to signal "private." Other code should not access it directly.

Why an ABC instead of a regular class?

A regular parent class would let you forget to implement send() in a subclass. Everything would look fine until runtime, when a missing method crashes your program. An ABC catches this mistake immediately at object creation.

  • Save channels.py.

You will not see a visible change yet. The next chunks build the concrete channels that bring this interface to life.

Implement ConsoleChannel and FileChannel

With the contract defined, you can now write classes that fulfil it. Each concrete channel inherits from NotificationChannel and provides its own implementation of send() and name. This is inheritance in action.

  • Add the ConsoleChannel class below the NotificationChannel class in channels.py:
class ConsoleChannel(NotificationChannel):
    # Map each priority to a readable label
    PRIORITY_LABELS = {
        Priority.LOW: "[LOW]",
        Priority.MEDIUM: "[MEDIUM]",
        Priority.HIGH: "[HIGH]",
        Priority.CRITICAL: "[CRITICAL]",
    }

    def __init__(self, min_priority: Priority = Priority.LOW):
        super().__init__(min_priority)

    @property
    def name(self) -> str:
        return "Console"

    def send(self, notification: Notification) -> bool:
        label = self.PRIORITY_LABELS.get(notification.priority, "[UNKNOWN]")
        timestamp = notification.timestamp.strftime("%Y-%m-%d %H:%M:%S")
        print(f"{timestamp} {label} {notification.message}")
        return True

What does this code do?

  • PRIORITY_LABELS is a class-level dictionary that maps each Priority enum value to a human-readable label for output formatting.
  • super().__init__(min_priority) calls the parent class constructor so _min_priority gets set properly.
  • send() formats the notification with a timestamp, priority label, and message, then prints it to the terminal. It returns True to signal success.
  • Now add the FileChannel class directly below ConsoleChannel:
class FileChannel(NotificationChannel):
    def __init__(self, filepath: str, min_priority: Priority = Priority.MEDIUM):
        super().__init__(min_priority)
        # Store the path where notifications will be logged
        self._filepath = filepath

    @property
    def name(self) -> str:
        return "File"

    def send(self, notification: Notification) -> bool:
        timestamp = notification.timestamp.strftime("%Y-%m-%d %H:%M:%S")
        line = f"{timestamp} [{notification.priority.name}] {notification.message}\n"
        with open(self._filepath, "a", encoding="utf-8") as f:
            f.write(line)
        return True

What does this code do?

  • FileChannel accepts a filepath in its constructor. This makes the log file location configurable per instance.
  • The default min_priority is MEDIUM. Low-priority messages skip the file by default, keeping log files focused on important events.
  • open(self._filepath, "a", encoding="utf-8") opens the file in append mode. Each notification adds a new line without overwriting previous entries.
  • Save channels.py.

You will not see a visible change yet. The next chunk adds the final channel, and then you will test all three together.

Seeing a red underline on the imports?

Cursor may show an import error for notification_engine.models. This happens when the editor has not reloaded the package structure. Save all files and the error should disappear.

If it persists, make sure your __init__.py file exists inside the notification_engine/ folder.

Help me debug this import error in my notification engine

Implement WebhookChannel

The third channel simulates sending an HTTP POST request. It does not actually make a network call. Instead, it prints the payload that would be sent. This demonstrates that the same interface works regardless of the underlying transport.

  • Add the WebhookChannel class below FileChannel in channels.py:
class WebhookChannel(NotificationChannel):
    def __init__(self, url: str, min_priority: Priority = Priority.CRITICAL):
        super().__init__(min_priority)
        # Store the endpoint URL for the simulated webhook
        self._url = url

    @property
    def name(self) -> str:
        return "Webhook"

    def send(self, notification: Notification) -> bool:
        payload = {
            "url": self._url,
            "priority": notification.priority.name,
            "message": notification.message,
            "timestamp": notification.timestamp.isoformat(),
        }
        print(f"[Webhook] Simulated POST to {self._url}")
        print(f"  Payload: {payload}")
        return True

What does this code do?

  • WebhookChannel takes a url parameter representing the endpoint it would POST to in production.
  • Its default min_priority is CRITICAL. Webhooks are expensive operations in real systems, so they only fire for the most urgent notifications.
  • send() builds a dictionary payload and prints it instead of making a real HTTP request. The structure is identical to what a real webhook integration would send.

This is polymorphism

All three channels have the same method signature: send(notification) -> bool. Any code that calls send() does not need to know whether it is talking to a console, a file, or a webhook. It just calls the method and trusts the contract.

This is polymorphism. One interface, many implementations. The router you build next takes full advantage of this.

  • Save channels.py.
  • Test all three channels by running this command in your integrated terminal:
python -c "
from notification_engine.models import Notification, Priority
from notification_engine.channels import ConsoleChannel, FileChannel, WebhookChannel

n = Notification(message='Server restarted', priority=Priority.CRITICAL)

console = ConsoleChannel()
console.send(n)

file_ch = FileChannel(filepath='test.log')
file_ch.send(n)

webhook = WebhookChannel(url='https://example.com/hook')
webhook.send(n)
"

You should see three outputs: the console prints a formatted timestamp with a [CRITICAL] label, the file channel confirms silently (check test.log to see the entry), and the webhook prints its simulated POST payload.

Seeing a ModuleNotFoundError?

Make sure your terminal is in the project root folder (the one containing the notification_engine/ directory). Python needs to see the package folder to resolve the import.

Also confirm that __init__.py exists inside notification_engine/. Without it, Python does not recognize the directory as a package.

Help me fix this ModuleNotFoundError for my notification_engine package

✔️ Awesome, I've got everything!

Great. Double check you have saved channels.py and you can clean up the test file by running:

rm test.log

ⓧ I'd like to double check the full code

Here is the full channels.py file for comparison:

from abc import ABC, abstractmethod

from notification_engine.models import Notification, Priority


class NotificationChannel(ABC):
    def __init__(self, min_priority: Priority = Priority.LOW):
        self._min_priority = min_priority

    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @abstractmethod
    def send(self, notification: Notification) -> bool:
        pass

    def supports_priority(self, notification: Notification) -> bool:
        return notification.priority >= self._min_priority


class ConsoleChannel(NotificationChannel):
    PRIORITY_LABELS = {
        Priority.LOW: "[LOW]",
        Priority.MEDIUM: "[MEDIUM]",
        Priority.HIGH: "[HIGH]",
        Priority.CRITICAL: "[CRITICAL]",
    }

    def __init__(self, min_priority: Priority = Priority.LOW):
        super().__init__(min_priority)

    @property
    def name(self) -> str:
        return "Console"

    def send(self, notification: Notification) -> bool:
        label = self.PRIORITY_LABELS.get(notification.priority, "[UNKNOWN]")
        timestamp = notification.timestamp.strftime("%Y-%m-%d %H:%M:%S")
        print(f"{timestamp} {label} {notification.message}")
        return True


class FileChannel(NotificationChannel):
    def __init__(self, filepath: str, min_priority: Priority = Priority.MEDIUM):
        super().__init__(min_priority)
        self._filepath = filepath

    @property
    def name(self) -> str:
        return "File"

    def send(self, notification: Notification) -> bool:
        timestamp = notification.timestamp.strftime("%Y-%m-%d %H:%M:%S")
        line = f"{timestamp} [{notification.priority.name}] {notification.message}\n"
        with open(self._filepath, "a", encoding="utf-8") as f:
            f.write(line)
        return True


class WebhookChannel(NotificationChannel):
    def __init__(self, url: str, min_priority: Priority = Priority.CRITICAL):
        super().__init__(min_priority)
        self._url = url

    @property
    def name(self) -> str:
        return "Webhook"

    def send(self, notification: Notification) -> bool:
        payload = {
            "url": self._url,
            "priority": notification.priority.name,
            "message": notification.message,
            "timestamp": notification.timestamp.isoformat(),
        }
        print(f"[Webhook] Simulated POST to {self._url}")
        print(f"  Payload: {payload}")
        return True

Three channels, one interface. Each channel handles notifications in its own way, but from the outside they all look identical. Next up, you will build the router that exploits this uniformity to dispatch notifications without knowing which channels are registered.

Build the Router with Composition and Polymorphism

You now have three channel classes that all follow the same abstract interface. The next challenge: how do you coordinate them so a single notification reaches the right destinations?

In this step, you'll build a NotificationRouter class that manages multiple channels and dispatches notifications to the right ones based on priority. The router uses composition (it holds channels) instead of inheritance (it does not extend them).

In this step, get ready to:
  • Define the NotificationRouter class with channel management methods.
  • Implement the dispatch method that routes notifications based on priority.
  • Add batch processing and a helpful string representation.
Define the router class with channel management

The router is the central coordinator of your notification system. It does not inherit from NotificationChannel. It holds a list of channels internally. This is the has-a relationship (composition) instead of the is-a relationship (inheritance).

  • Open notification_engine/router.py in Cursor and add the following code:
from notification_engine.models import Notification
from notification_engine.channels import NotificationChannel


class NotificationRouter:
    def __init__(self):
        # Private list holding all registered channels
        self._channels: list[NotificationChannel] = []

    def register_channel(self, channel: NotificationChannel) -> None:
        # Add a channel to the routing pool
        self._channels.append(channel)

    def remove_channel(self, name: str) -> None:
        # Filter out the channel matching the given name
        self._channels = [ch for ch in self._channels if ch.name != name]

What does this code do?

  • _channels is a private list (prefixed with underscore) that stores all registered channels. The type hint list[NotificationChannel] means it can hold any object that implements the abstract interface.
  • register_channel() appends a channel to the list. remove_channel() filters out any channel matching the given name using a list comprehension.
  • The router's type hint accepts NotificationChannel (the abstract base class), so any concrete subclass (Console, File, Webhook) can be registered without the router knowing its specific type.

Why composition instead of inheritance here?

The router manages channels. It is not a channel itself. If you made it inherit from NotificationChannel, you'd be saying "a router IS a channel" which makes no sense.

Composition (has-a) means the router HAS channels. This is a cleaner model because you can add or remove channels at runtime without changing the router's own class hierarchy.

Implement the dispatch method

The dispatch method is where polymorphism comes alive. The router calls send() on each qualifying channel without knowing or caring which concrete type it is.

  • In notification_engine/router.py, add the dispatch method inside the NotificationRouter class, below remove_channel:
    def dispatch(self, notification: Notification) -> dict:
        # Track which channels handled, skipped, or failed the notification
        sent_to = []
        skipped = []
        failed = []

        for channel in self._channels:
            # Check if this channel accepts the notification's priority level
            if not channel.supports_priority(notification):
                skipped.append(channel.name)
                continue
            try:
                # Polymorphism: call send() regardless of concrete channel type
                success = channel.send(notification)
                if success:
                    sent_to.append(channel.name)
                else:
                    failed.append(channel.name)
            except Exception as e:
                failed.append(f"{channel.name} (error: {e})")

        return {"sent_to": sent_to, "skipped": skipped, "failed": failed}

What does this code do?

  • The method loops through every registered channel and first checks supports_priority(). If the notification's priority is below the channel's minimum, that channel is skipped.
  • For qualifying channels, the router calls send() inside a try/except block. If a single channel throws an error, the others still get their turn.
  • The return dict gives the caller a complete summary of what happened. This is useful for logging, debugging, or displaying results.
Add batch dispatch and a string representation
  • Still in notification_engine/router.py, add these two methods below dispatch:
    def dispatch_batch(self, notifications: list[Notification]) -> list[dict]:
        # Process multiple notifications and collect results for each
        return [self.dispatch(n) for n in notifications]

    def __repr__(self) -> str:
        return f"NotificationRouter(channels={len(self._channels)})"

What does this code do?

  • dispatch_batch() is a convenience method that dispatches a list of notifications in one call. It reuses dispatch() for each one and collects the results.
  • __repr__ gives you a readable string when you print the router (e.g. NotificationRouter(channels=3)). This is helpful during debugging.
  • Save notification_engine/router.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on to the test.

ⓧ I'd like to double check the full code

from notification_engine.models import Notification
from notification_engine.channels import NotificationChannel


class NotificationRouter:
    def __init__(self):
        self._channels: list[NotificationChannel] = []

    def register_channel(self, channel: NotificationChannel) -> None:
        self._channels.append(channel)

    def remove_channel(self, name: str) -> None:
        self._channels = [ch for ch in self._channels if ch.name != name]

    def dispatch(self, notification: Notification) -> dict:
        sent_to = []
        skipped = []
        failed = []

        for channel in self._channels:
            if not channel.supports_priority(notification):
                skipped.append(channel.name)
                continue
            try:
                success = channel.send(notification)
                if success:
                    sent_to.append(channel.name)
                else:
                    failed.append(channel.name)
            except Exception as e:
                failed.append(f"{channel.name} (error: {e})")

        return {"sent_to": sent_to, "skipped": skipped, "failed": failed}

    def dispatch_batch(self, notifications: list[Notification]) -> list[dict]:
        return [self.dispatch(n) for n in notifications]

    def __repr__(self) -> str:
        return f"NotificationRouter(channels={len(self._channels)})"
  • Test the router by running this command in your terminal:
python -c "
from notification_engine import *
router = NotificationRouter()
router.register_channel(ConsoleChannel(min_priority=Priority.LOW))
router.register_channel(FileChannel(filepath='test.log', min_priority=Priority.MEDIUM))
router.register_channel(WebhookChannel(url='https://example.com/hook', min_priority=Priority.CRITICAL))
n = Notification(message='Server overloaded', priority=Priority.HIGH)
result = router.dispatch(n)
print(result)
"

You should see Console print the formatted notification, followed by the result dict showing Console and File in sent_to and Webhook in skipped.

Why was Webhook skipped?

The Webhook channel has a minimum priority of CRITICAL (value 4). The test notification has priority HIGH (value 3). Since HIGH is less than CRITICAL, supports_priority() returns False and the router skips it.

Seeing a ModuleNotFoundError?

  • Make sure you're running the command from your project's root directory (the folder containing notification_engine/).
  • Check that __init__.py still imports NotificationRouter from notification_engine.router.
  • Confirm that router.py is saved with the complete class definition.

Help me debug a ModuleNotFoundError when importing notification_engine

  • Delete the test.log file that was created by the FileChannel during your test:
rm test.log

Your routing engine is fully wired. The router dispatches to any channel without knowing its concrete type. Next up, you'll bring everything together in main.py and watch the full system handle notifications across all four priority levels.

Wire It All Together in main.py

You now have all the individual pieces: a data model, three channel implementations, and a router that dispatches polymorphically. The missing piece is a CLI entry point that demonstrates the full system working together.

In this step, you will build main.py to instantiate channels with different priority thresholds, feed in test notifications, and observe routing in action. This is where the Open/Closed Principle pays off.

In this step, get ready to:
  • Import from the package and instantiate three channels with different priority thresholds.
  • Create test notifications spanning all four priority levels and dispatch them.
  • Add error handling and a final summary showing per-channel dispatch counts.
Import from the package and instantiate the channels

The __init__.py you set up earlier exposes all the public classes at the package level. This means main.py can import everything it needs from a single location.

  • In Cursor, open main.py (the file at the top level of your project, outside the notification_engine/ folder).
  • Replace the empty file with the following code:
from notification_engine import (
    Notification,
    Priority,
    ConsoleChannel,
    FileChannel,
    WebhookChannel,
    NotificationRouter,
)


def main():
    # Instantiate channels with different minimum priority thresholds
    console = ConsoleChannel(min_priority=Priority.LOW)
    file_channel = FileChannel(filepath="notifications.log", min_priority=Priority.MEDIUM)
    webhook = WebhookChannel(url="https://example.com/webhook", min_priority=Priority.CRITICAL)

    # Create the router and register all channels
    router = NotificationRouter()
    router.register_channel(console)
    router.register_channel(file_channel)
    router.register_channel(webhook)

    print(f"Router ready: {router}")
    print("=" * 60)

What does this code do?

Each channel is created with a different min_priority threshold. This controls which notifications each channel accepts.

  • ConsoleChannel accepts everything (LOW and above).
  • FileChannel only logs MEDIUM priority and above to notifications.log.
  • WebhookChannel only fires for CRITICAL notifications.

The router uses composition to manage these channels. It does not know or care what concrete type each channel is. It only knows they all satisfy the NotificationChannel interface.

  • Save main.py.

Seeing a red underline on the import?

Make sure your terminal is in the project root folder (the one containing both main.py and the notification_engine/ directory). Cursor's language server resolves imports relative to where the terminal is opened.

Also confirm that __init__.py exists inside notification_engine/. Without it, Python does not recognize the directory as a package.

Help me debug my import error in main.py

Create test notifications and dispatch them

To see routing in action, you need notifications at every priority level. Each one will take a different path through the system based on the channel thresholds you just configured.

  • Add the following code below the print("=" * 60) line, still inside the main() function:
    # Test notifications spanning all four priority levels
    notifications = [
        Notification(message="System health check passed.", priority=Priority.LOW),
        Notification(message="Disk usage at 80% on server-3.", priority=Priority.MEDIUM),
        Notification(message="API response time exceeding 2s threshold.", priority=Priority.HIGH),
        Notification(message="Database primary node unreachable!", priority=Priority.CRITICAL),
    ]

What does this code do?

Each notification is created with the Notification dataclass. The timestamp field auto-populates via the default_factory you defined in models.py.

The four messages simulate realistic monitoring scenarios. The LOW notification will only reach the console. The CRITICAL one will reach all three channels.

  • Save main.py.
Add the dispatch loop with error handling and summary

The dispatch loop iterates through every notification and sends it to the router. Wrapping each dispatch in a try/except ensures that one failing channel does not crash the entire system.

  • Add the following code below the notifications list, still inside the main() function:
    total_sent = 0
    channel_counts: dict[str, int] = {}

    for notification in notifications:
        print(f"\nDispatching: [{notification.priority.name}] {notification.message}")
        print("-" * 40)

        # Error handling ensures one failure does not crash the system
        try:
            result = router.dispatch(notification)
        except Exception as e:
            print(f"  DISPATCH ERROR: {e}")
            continue

        # Report which channels handled this notification
        if result["sent_to"]:
            print(f"  Sent to: {', '.join(result['sent_to'])}")
            total_sent += 1
            for ch_name in result["sent_to"]:
                channel_counts[ch_name] = channel_counts.get(ch_name, 0) + 1
        if result["skipped"]:
            print(f"  Skipped: {', '.join(result['skipped'])}")
        if result["failed"]:
            print(f"  Failed: {', '.join(result['failed'])}")

    # Final summary
    print("\n" + "=" * 60)
    print(f"Summary: {total_sent} notifications dispatched successfully.")
    for ch_name, count in channel_counts.items():
        print(f"  {ch_name}: {count} notifications")


if __name__ == "__main__":
    main()

What does this code do?

  • The try/except around router.dispatch() catches any unexpected error from a channel without stopping the loop.
  • The channel_counts dictionary tracks how many notifications each channel handled, building a per-channel breakdown.
  • The if __name__ == "__main__" guard ensures main() only runs when the file is executed directly, not when imported as a module.

Why is this the Open/Closed Principle?

If you wanted to add an EmailChannel tomorrow, you would write one new class that inherits from NotificationChannel and register it with the router. Zero changes to the router. Zero changes to existing channels. The system is open for extension but closed for modification.

  • Save main.py.
  • Run the engine by entering this command in your integrated terminal:
python main.py
  • You should see formatted output showing each notification being dispatched with routing summaries. The LOW notification goes to Console only. The CRITICAL notification reaches all three channels, including a simulated webhook payload.
  • Confirm that a notifications.log file was created in your project folder by running:
cat notifications.log
  • You should see three entries (MEDIUM, HIGH, and CRITICAL) formatted with timestamps and priority labels. The LOW notification is absent because FileChannel has a minimum priority of MEDIUM.

Seeing a ModuleNotFoundError?

Make sure your terminal's working directory is the project root (the folder containing both main.py and notification_engine/). If you are inside the notification_engine/ folder, Python cannot resolve the package import.

Help me fix my ModuleNotFoundError when running main.py

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and confirmed both the terminal output and the notifications.log file.

ⓧ I'd like to double check the full code

Here is the complete main.py file:

from notification_engine import (
    Notification,
    Priority,
    ConsoleChannel,
    FileChannel,
    WebhookChannel,
    NotificationRouter,
)


def main():
    console = ConsoleChannel(min_priority=Priority.LOW)
    file_channel = FileChannel(filepath="notifications.log", min_priority=Priority.MEDIUM)
    webhook = WebhookChannel(url="https://example.com/webhook", min_priority=Priority.CRITICAL)

    router = NotificationRouter()
    router.register_channel(console)
    router.register_channel(file_channel)
    router.register_channel(webhook)

    print(f"Router ready: {router}")
    print("=" * 60)

    notifications = [
        Notification(message="System health check passed.", priority=Priority.LOW),
        Notification(message="Disk usage at 80% on server-3.", priority=Priority.MEDIUM),
        Notification(message="API response time exceeding 2s threshold.", priority=Priority.HIGH),
        Notification(message="Database primary node unreachable!", priority=Priority.CRITICAL),
    ]

    total_sent = 0
    channel_counts: dict[str, int] = {}

    for notification in notifications:
        print(f"\nDispatching: [{notification.priority.name}] {notification.message}")
        print("-" * 40)

        try:
            result = router.dispatch(notification)
        except Exception as e:
            print(f"  DISPATCH ERROR: {e}")
            continue

        if result["sent_to"]:
            print(f"  Sent to: {', '.join(result['sent_to'])}")
            total_sent += 1
            for ch_name in result["sent_to"]:
                channel_counts[ch_name] = channel_counts.get(ch_name, 0) + 1
        if result["skipped"]:
            print(f"  Skipped: {', '.join(result['skipped'])}")
        if result["failed"]:
            print(f"  Failed: {', '.join(result['failed'])}")

    print("\n" + "=" * 60)
    print(f"Summary: {total_sent} notifications dispatched successfully.")
    for ch_name, count in channel_counts.items():
        print(f"  {ch_name}: {count} notifications")


if __name__ == "__main__":
    main()

Your notification routing engine is fully operational. You have a clean, extensible system where adding a new channel means writing one class and calling register_channel(). Next up, the Secret Mission challenges you to add a filtering system that gives channels even more control over which notifications they accept.

Secret mission

Add a FilterChain

Your routing engine dispatches by priority, but real systems need finer control. What if a channel should reject messages containing certain keywords? Or only accept notifications during business hours? Add a pluggable filter system that gives each channel custom accept/reject rules.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources for future reference or delete them entirely.

Resources you used:

  • Project folder on your Desktop containing all Python source files (local, no cost).
  • notifications.log file generated by running the program (local, no cost).

Keep everything running

No action needed. Choose this if you plan to extend the notification engine further or use it as reference for future OOP projects.

  • Your project folder stays on your Desktop with all source files intact.
  • The notifications.log file remains in the project folder and will be appended to each time you run main.py.

Pause - I'll come back to this later

Close the project in Cursor to free up workspace. Your files stay on disk and you can reopen them any time.

  • In Cursor, click File, then Close Folder.
  • To reopen later, click File, then Open Folder, and navigate to your project folder on the Desktop.

Delete - I don't want to use this again

Remove all project files and start fresh. This cannot be undone.

  • Close the project in Cursor first.
  • Delete the notifications.log file and the entire project folder by running these commands in your terminal:

macOS

cd ~/Desktop
rm -rf notification-engine

Windows

cd ~\Desktop
Remove-Item -Recurse -Force notification-engine
  • Confirm the folder is gone by running ls (macOS) or dir (Windows) on your Desktop. You should no longer see the project folder listed.

Nice Work!

Nice Work!

You just built a complete notification routing engine from scratch using object-oriented programming principles in Python. This is the same architectural pattern used in production logging frameworks, event systems, and pub/sub architectures.

You've learned how to:

  • Define abstract interfaces using Python's ABC module that enforce contracts on subclasses, preventing instantiation until all abstract methods are implemented.
  • Implement polymorphism so the router dispatches notifications to any channel without knowing or caring about its concrete type.
  • Apply the Open/Closed Principle and composition over inheritance so that adding a new notification channel requires zero changes to existing code.
  • Secret Mission: Implemented the Chain of Responsibility pattern with a pluggable filter system that lets channels accept or reject notifications based on custom rules.

Ready to quiz yourself?