Skip to content

A production-ready notification service built with Python that supports multiple channels, templating, and event-driven architecture

Notifications You must be signed in to change notification settings

saggit/notify_demo

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

1 Commit
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Notification Service

A production-ready notification service built with Python that supports multiple channels, templating, and event-driven architecture.

Features

  • Multiple Notification Channels: Email, Slack, and extensible for more
  • Template Engine: Jinja2-based templating for dynamic message content
  • Event Sources: Real-time and scheduled event processing
  • Deduplication: Configurable deduplication policies to prevent spam
  • Modular Architecture: Clean separation of concerns with dependency injection
  • Production Ready: Proper logging, configuration management, and error handling

Project Structure

upwork/
├── src/                    # Source code package
│   ├── __init__.py        # Package initialization
│   ├── channels.py        # Notification channel implementations
│   ├── templates.py       # Template rendering engine
│   ├── templates_data.py  # Template definitions
│   ├── events.py          # Event source implementations
│   ├── deduplication.py   # Deduplication policies
│   ├── registry.py        # Notification configuration registry
│   ├── service.py         # Main notification service
│   └── config.py          # Application configuration
├── main.py                # Application entry point
├── pyproject.toml         # Project dependencies
└── README.md             # This file

Quick Start

Prerequisites

  • Python 3.12+
  • uv (Python package manager)

Installation

  1. Clone the repository
  2. Install dependencies:
    uv sync

Running the Service

uv run python main.py

This will:

  • Process user signup events
  • Send welcome emails and Slack notifications
  • Generate daily statistics reports

Architecture

Core Components

1. Notification Channels (src/channels.py)

  • EmailChannel: Writes notifications to log files
  • SlackChannel: Prints notifications to console (mock implementation)
  • Extensible: Add new channels by implementing NotificationChannel

2. Template Engine (src/templates.py)

  • Jinja2-based template rendering
  • Supports dynamic content with context variables
  • Templates stored in src/templates_data.py

3. Event Sources (src/events.py)

  • RealTimeSource: Processes immediate events
  • ScheduledSource: Queries databases for scheduled reports
  • Extensible: Add new sources by implementing EventSource

4. Deduplication (src/deduplication.py)

  • SimpleDedupPolicy: In-memory deduplication by event ID
  • Prevents duplicate notifications
  • Each notification config can have its own policy

5. Service Layer (src/service.py)

  • Orchestrates the entire notification workflow
  • Processes events, renders templates, and sends notifications

Configuration

Environment variables can be set to configure the service:

export EMAIL_OUTPUT_DIR="./logs"
export DEFAULT_SLACK_CHANNEL="#alerts"
export ENABLE_EMAIL=true
export ENABLE_SLACK=true

Adding New Features

Adding a New Channel

  1. Implement NotificationChannel in src/channels.py:

    class DiscordChannel(NotificationChannel):
        def __init__(self, webhook_url: str):
            self.webhook_url = webhook_url
        
        def send(self, message: str) -> None:
            # Implementation here
            pass
  2. Register in your notification configuration:

    discord_channel = DiscordChannel(webhook_url="https://...")
    registry.register(NotificationConfig(
        name='Discord Alert',
        channel=discord_channel,
        template_name='alert_discord.txt',
        event_source=event_source,
        dedup_policy=SimpleDedupPolicy(),
    ))

Adding a New Template

  1. Add template to src/templates_data.py:

    TEMPLATES = {
        # ... existing templates
        'alert_discord.txt': "🚨 {{ message }} - {{ timestamp }}",
    }
  2. Reference in notification configuration:

    template_name='alert_discord.txt'

Adding a New Event Source

  1. Implement EventSource in src/events.py:

    class APISource(EventSource):
        def __init__(self, api_url: str):
            self.api_url = api_url
        
        def fetch(self) -> List[Dict[str, Any]]:
            # Fetch from API
            return events
  2. Use in notification configuration:

    api_source = APISource(api_url="https://api.example.com/events")

Development

Running Tests

uv run pytest

Code Style

This project follows PEP 8 and uses type hints throughout.

Adding Dependencies

uv add package-name

Production Deployment

Docker

FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install uv && uv sync --frozen
CMD ["uv", "run", "python", "main.py"]

Environment Configuration

Set up environment variables for production:

export DATABASE_URL="postgresql://user:pass@host/db"
export LOG_LEVEL="INFO"
export EMAIL_OUTPUT_DIR="/var/log/notifications"

Contributing

  1. Fork the repository
  2. Create a feature branch
  3. Make your changes
  4. Add tests if applicable
  5. Submit a pull request

License

MIT License - see LICENSE file for details.

About

A production-ready notification service built with Python that supports multiple channels, templating, and event-driven architecture

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published

Languages