Build a Notification Service from Scratch — Part 2: Queue and Worker System

# app/celery_app.py
from celery import Celery
from app.config import settings

celery_app = Celery(
    "notification_service",
    broker=settings.redis_url,
    backend=settings.redis_url
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,           # Re-deliver if worker crashes
    task_reject_on_worker_lost=True,
    task_default_retry_delay=60,    # 1 minute
    task_max_retries=3,
    worker_prefetch_multiplier=1     # One task at a time per worker
)

# app/queue.py
from app.celery_app import celery_app
from app.models import NotificationStatus

@celery_app.task(bind=True, max_retries=3)
def deliver_notification(self, notification_id: int):
    from app.core.database import SessionLocal
    from app.models import Notification

    db = SessionLocal()
    notification = db.query(Notification).get(notification_id)

    if not notification:
        return

    try:
        # Dispatch based on channel
        if notification.channel == "email":
            send_email(notification.recipient, notification.subject, notification.body)
        elif notification.channel == "sms":
            send_sms(notification.recipient, notification.body)
        elif notification.channel == "push":
            send_push(notification.recipient, notification.body)

        notification.status = NotificationStatus.SENT
        notification.sent_at = datetime.now(timezone.utc)
        db.commit()

    except Exception as e:
        notification.retry_count += 1
        db.commit()

        if notification.retry_count < notification.max_retries:
            raise self.retry(exc=e, countdown=60 * (2 ** notification.retry_count))
        else:
            notification.status = NotificationStatus.FAILED
            db.commit()
    finally:
        db.close()

def enqueue_notification(notification_id: int):
    deliver_notification.delay(notification_id)

# app/dispatchers/email.py
import smtplib
from email.mime.text import MIMEText
from app.config import settings

def send_email(to: str, subject: str, body: str):
    msg = MIMEText(body, "html")
    msg["Subject"] = subject
    msg["From"] = settings.email_from
    msg["To"] = to

    with smtplib.SMTP(settings.smtp_host, settings.smtp_port) as server:
        server.starttls()
        server.login(settings.smtp_user, settings.smtp_password)
        server.send_message(msg)

# app/dispatchers/sms.py
import requests

def send_sms(to: str, body: str):
    """Send SMS via Twilio (or any provider)."""
    response = requests.post(
        f"https://api.twilio.com/2010-04-01/Accounts/{settings.twilio_sid}/Messages.json",
        auth=(settings.twilio_sid, settings.twilio_token),
        data={"To": to, "From": settings.twilio_phone, "Body": body}
    )
    response.raise_for_status()

# app/dispatchers/push.py
def send_push(device_token: str, body: str):
    """Send push notification via FCM."""
    requests.post(
        "https://fcm.googleapis.com/fcm/send",
        headers={"Authorization": f"key={settings.fcm_key}"},
        json={"to": device_token, "notification": {"body": body}}
    )

# app/api/notifications.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from app.services.notification_service import NotificationService
from app.core.database import get_db

router = APIRouter(prefix="/notifications", tags=["notifications"])

class SendRequest(BaseModel):
    template_name: str
    recipient: str
    context: dict = {}

@router.post("/send", status_code=202)
def send_notification(req: SendRequest, db=Depends(get_db)):
    service = NotificationService(db)
    notification = service.send(req.template_name, req.recipient, req.context)
    return {"id": notification.id, "status": notification.status}

@router.get("/{notification_id}")
def get_status(notification_id: int, db=Depends(get_db)):
    notification = db.query(Notification).get(notification_id)
    if not notification:
        raise HTTPException(404, "Notification not found")
    return {"id": notification.id, "status": notification.status, "retry_count": notification.retry_count}

# Start a Celery worker
celery -A app.celery_app worker --loglevel=info --concurrency=4

# Start the API
uvicorn app.main:app --reload

# Create a template first
curl -X POST http://localhost:8000/templates \
  -H "Content-Type: application/json" \
  -d '{"name":"welcome","channel":"email","subject_template":"Welcome!","body_template":"Hello {{user_name}}!","variables":["user_name"]}'

# Send a notification
curl -X POST http://localhost:8000/notifications/send \
  -H "Content-Type: application/json" \
  -d '{"template_name":"welcome","recipient":"user@example.com","context":{"user_name":"John"}}'

# Check status
curl http://localhost:8000/notifications/1
# {"id": 1, "status": "sent", "retry_count": 0}