Task 04 — Add a New Alert Type

Difficulty: Intermediate
Goal: Add a new alert category to the AquaGen alert system end-to-end — from the enum definition through the processor and notification channels.

---

What you'll learn

• How the alert system is structured (alerts_processor.py, alerts_content/, channels/)
• How to define an alert with threshold logic
• How alerts are dispatched to Email / SMS / Chat channels

---

Background

Alerts in AquaGen follow a pipeline:

POST /api/user/alerts
    → AlertsProcessor.process_alerts()
        → get_content()         # maps alertType string → content builder class
        → get_recipient_contacts()  # resolves email, SMS, FCM, web recipients
        → send_to_all_channel() # dispatches: email → SMS → WhatsApp → FCM → Web

Each alert type has:

1. A constant in app/services/alerts/constants/
2. A content builder in app/services/alerts/alerts_content/
3. A branch in AlertsProcessor.get_content() that maps the alertType string to the content class

See Adding a New Alert Type for the full step-by-step guide.

---

Your task: "High Conductivity Alert"

Add an alert that fires when a quality sensor reports conductivity above a configurable threshold.

Suggested alert name: HIGH_CONDUCTIVITY

---

Files to create / edit

File	What to do
app/services/alerts/constants/alert_types.py	Add HIGH_CONDUCTIVITY = 'HIGH_CONDUCTIVITY'
app/services/alerts/alerts_content/high_conductivity_content.py	Build the alert message
app/services/alerts/alerts_processor.py	Add evaluation branch + channel dispatch

---

Content builder skeleton

# app/services/alerts/alerts_content/high_conductivity_content.py

class HighConductivityContent:
    def __init__(self, unit_id, unit_name, value, threshold, industry_name):
        self.unit_id       = unit_id
        self.unit_name     = unit_name
        self.value         = value
        self.threshold     = threshold
        self.industry_name = industry_name

    def build(self):
        return {
            'subject': f'High Conductivity Alert — {self.unit_name}',
            'body': (
                f'Conductivity at {self.unit_name} has reached {self.value} µS/cm, '
                f'exceeding the threshold of {self.threshold} µS/cm.'
            ),
            'unit_id':       self.unit_id,
            'industry_name': self.industry_name,
            'alert_type':    'HIGH_CONDUCTIVITY',
        }

Processor branch

Inside AlertsProcessor.get_content(), add an elif branch that maps the alertType string to your content builder:

from app.services.alerts.alerts_content.high_conductivity_content import HighConductivityContent

# Inside get_content() — alongside all other elif branches:
elif self.request.alert == 'HIGH_CONDUCTIVITY':
    return HighConductivityContent(
        unit_id=self.request.unit_id,
        unit_name=self.unit_name,
        value=self.request.data.get('meta', {}).get('currentValue'),
        threshold=self.request.data.get('meta', {}).get('threshold', 1500),
        industry_name=self.request.industry_name,
    ).build()

The processor automatically calls send_to_all_channel() with the returned content dict — you don't need to dispatch manually.

---

Testing the alert

1. Set a low conductivityThreshold on a test unit in the DB.
2. Insert a raw sensor reading above that threshold.
3. Trigger the alert evaluation manually (or hit the relevant endpoint).
4. Verify the notification was queued / sent.

---

Done when

• HIGH_CONDUCTIVITY constant exists
• Content builder produces a correctly shaped dict
• Processor fires when the threshold is breached and skips when it isn't
• At least one channel (Email or SMS) receives the alert message in test