Build a Notification Service from Scratch — Part 1: Architecture and Database Design

Series Overview

We’re building a notification service that sends messages across multiple channels (email, SMS, push). It handles templating, queuing, retries, and rate limiting — designed to be a standalone microservice.

Technology stack: FastAPI, PostgreSQL, Redis, Celery.

Architecture

┌──────────┐     ┌──────────────┐     ┌──────────┐     ┌──────────┐
│  Your    │────▶│ Notification │────▶│  Redis   │────▶│  Celery  │
│  App     │◀────│   Service    │◀────│  Queue   │◀────│  Workers │
└──────────┘     └──────┬───────┘     └──────────┘     └─────┬────┘
                        │                                     │
                        ▼                                     ▼
                 ┌──────────┐                          ┌──────────┐
                 │PostgreSQL│                          │  Email   │
                 │ (DB)     │                          │  SMS     │
                 └──────────┘                          │  Push    │
                                                       └──────────┘

Database Schema

# app/models.py
from sqlalchemy import Column, BigInteger, String, Text, DateTime, JSON, ForeignKey, Enum
from sqlalchemy.orm import declarative_base
import enum

Base = declarative_base()

class NotificationStatus(str, enum.Enum):
    PENDING = "pending"
    QUEUED = "queued"
    SENT = "sent"
    FAILED = "failed"
    RETRYING = "retrying"

class NotificationChannel(str, enum.Enum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"

class NotificationTemplate(Base):
    __tablename__ = "notification_templates"

    id = Column(BigInteger, primary_key=True)
    name = Column(String(100), unique=True, nullable=False)
    channel = Column(Enum(NotificationChannel), nullable=False)
    subject_template = Column(Text, nullable=True)  # For email
    body_template = Column(Text, nullable=False)
    variables = Column(JSON, default=list)  # ["user_name", "order_id"]

class Notification(Base):
    __tablename__ = "notifications"

    id = Column(BigInteger, primary_key=True)
    template_id = Column(BigInteger, ForeignKey("notification_templates.id"))
    channel = Column(Enum(NotificationChannel), nullable=False)
    recipient = Column(String(255), nullable=False)  # email/phone/device_token
    subject = Column(Text, nullable=True)
    body = Column(Text, nullable=False)
    context = Column(JSON, default=dict)  # {"user_name": "John"}
    status = Column(Enum(NotificationStatus), default=NotificationStatus.PENDING)
    retry_count = Column(Integer, default=0)
    max_retries = Column(Integer, default=3)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    sent_at = Column(DateTime, nullable=True)

Template Engine

# app/template_engine.py
from jinja2 import Template

def render_template(template_str: str, context: dict) -> str:
    """Render a Jinja2 template with variables."""
    return Template(template_str).render(**context)

# Example usage:
body_template = "Hello {{ user_name }}, your order #{{ order_id }} is confirmed."
context = {"user_name": "John", "order_id": "12345"}
result = render_template(body_template, context)
# "Hello John, your order #12345 is confirmed."

Creating Templates and Notifications

# app/services/notification_service.py
class NotificationService:
    def __init__(self, db: Session):
        self.db = db

    def send(self, template_name: str, recipient: str, context: dict):
        template = self.db.query(NotificationTemplate).filter_by(name=template_name).first()
        if not template:
            raise ValueError(f"Template '{template_name}' not found")

        body = render_template(template.body_template, context)
        subject = render_template(template.subject_template, context) if template.subject_template else None

        notification = Notification(
            template_id=template.id,
            channel=template.channel,
            recipient=recipient,
            subject=subject,
            body=body,
            context=context
        )
        self.db.add(notification)
        self.db.commit()

        # Push to queue
        enqueue_notification(notification.id)
        return notification

Summary

We’ve designed the database schema for multi-channel notifications with Jinja2 templating and a service layer that creates notifications and enqueues them for delivery.

Part 2 →

Advertisement