A production-ready notification service built with Python that supports multiple channels, templating, and event-driven architecture.
- 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
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
- Python 3.12+
- uv (Python package manager)
- Clone the repository
- Install dependencies:
uv sync
uv run python main.pyThis will:
- Process user signup events
- Send welcome emails and Slack notifications
- Generate daily statistics reports
- EmailChannel: Writes notifications to log files
- SlackChannel: Prints notifications to console (mock implementation)
- Extensible: Add new channels by implementing
NotificationChannel
- Jinja2-based template rendering
- Supports dynamic content with context variables
- Templates stored in
src/templates_data.py
- RealTimeSource: Processes immediate events
- ScheduledSource: Queries databases for scheduled reports
- Extensible: Add new sources by implementing
EventSource
- SimpleDedupPolicy: In-memory deduplication by event ID
- Prevents duplicate notifications
- Each notification config can have its own policy
- Orchestrates the entire notification workflow
- Processes events, renders templates, and sends notifications
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-
Implement
NotificationChannelinsrc/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
-
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(), ))
-
Add template to
src/templates_data.py:TEMPLATES = { # ... existing templates 'alert_discord.txt': "🚨 {{ message }} - {{ timestamp }}", }
-
Reference in notification configuration:
template_name='alert_discord.txt'
-
Implement
EventSourceinsrc/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
-
Use in notification configuration:
api_source = APISource(api_url="https://api.example.com/events")
uv run pytestThis project follows PEP 8 and uses type hints throughout.
uv add package-nameFROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install uv && uv sync --frozen
CMD ["uv", "run", "python", "main.py"]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"- Fork the repository
- Create a feature branch
- Make your changes
- Add tests if applicable
- Submit a pull request
MIT License - see LICENSE file for details.