mesonet-alerts 0.1.0

Shared email alerting, retries, and volume-drop checks for Mesonet workers

Mesonet Alerts Package

A production-ready shared Python package for email alerting across mesonet microservices. Provides HTML + plaintext email templates, SMTP configuration, DynamoDB persistence, retry helpers, and volume drop detection.

🚀 Quick Start

Installation

Add to your service's pyproject.toml:

[project]
dependencies = [
    "mesonet_alerts @ file://../common/mesonet_alerts"
]

Environment Variables

# SMTP Configuration (required for email sending)
ALERTS_SMTP_HOST=localhost
ALERTS_SMTP_PORT=1025
ALERTS_SMTP_USER=                    # Optional for local dev
ALERTS_SMTP_PASS=                    # Optional for local dev
ALERTS_FROM="[e-mail]"
ALERTS_TO="[e-mail],[e-mail]"

# Optional persistence
ALERTS_TABLE_NAME=alerts             # Enable DynamoDB persistence

# Volume monitoring
EXPECTED_RECORDS_PER_PROVIDER_PER_HOUR=100

Basic Usage

from mesonet_alerts import EmailAlerter, AlertStore, run_with_retries

# Initialize components
emailer = EmailAlerter()
store = AlertStore()

# Send an alert
context = {
    "stage": "ingest",
    "severity": "ERROR", 
    "provider": "colorado",
    "run_id": "run_123",
    "error": "Connection timeout"
}

emailer.send("process_failure", "Alert: Colorado Ingest Failed", context)

# Store alert with deduplication
store.put_alert(
    provider="colorado",
    stage="ingest", 
    severity="ERROR",
    code="CONNECTION_TIMEOUT",
    message="Failed to connect to provider",
    dedupe_key="timeout#colorado#run_123"
)

📧 Email Templates

The package includes four pre-built templates with eye-catching HTML + plaintext versions:

• process_failure - General processing failures
• provider_empty_data - Empty data warnings
• harmonize_failure - Data harmonization errors
• volume_drop - Volume drop alerts

Template Variables

All templates support these variables:

context = {
    "stage": "ingest",           # Processing stage
    "severity": "ERROR",         # ERROR, WARN, INFO
    "provider": "colorado",      # Provider name
    "run_id": "run_123",        # Optional run identifier
    "trace_id": "trace_456",    # Optional trace identifier
    "error": "Error message",    # Error details
    "attempts": 3,              # Number of attempts
    "timestamp_iso": "2025-01-15T10:30:00Z",
    
    # Volume drop specific
    "expected": 100,            # Expected record count
    "actual": 75,              # Actual record count  
    "drop_pct": "25.0",        # Drop percentage
    "window_start": "2025-01-15 13:00 UTC",
    "window_end": "2025-01-15 14:00 UTC",
}

🔄 Retry Helpers

Basic Retry Usage

from mesonet_alerts.retry import run_with_retries, ProviderEmptyDataError

def fetch_provider_data():
    # Your data fetching logic
    data = api_client.get_data()
    if not data:
        raise ProviderEmptyDataError("No data returned")
    return data

def is_retryable_error(e):
    return isinstance(e, (ProviderEmptyDataError, ConnectionError))

try:
    data = run_with_retries(fetch_provider_data, is_retryable_error, attempts=3)
except ProviderEmptyDataError:
    # Handle final failure after retries
    pass

Decorator Usage

from mesonet_alerts.retry import retry_on_exceptions

@retry_on_exceptions(ProviderEmptyDataError, ConnectionError)
def fetch_with_auto_retry():
    return api_client.get_data()

📊 Volume Drop Detection

from datetime import datetime, timezone, timedelta
from mesonet_alerts.dropcheck import check_and_alert_volume_drop

# Define time window
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
window_end = now
window_start = now - timedelta(hours=1)

# Check for volume drops
check_and_alert_volume_drop(
    provider="colorado",
    stage="harmonize",
    actual_count=75,        # Only 75 records processed
    expected_count=100,     # Expected 100 records
    threshold=0.20,         # Alert on >20% drop
    window_start=window_start,
    window_end=window_end,
    emailer=emailer,
    store=store
)

🏗️ Integration Examples

Ingest Worker Integration

from mesonet_alerts.emailer import EmailAlerter
from mesonet_alerts.store import AlertStore
from mesonet_alerts.retry import run_with_retries, ProviderEmptyDataError

emailer = EmailAlerter()
store = AlertStore()

def _is_retryable(e: Exception) -> bool:
    return isinstance(e, ProviderEmptyDataError) or "timeout" in str(e).lower()

def fetch_and_process_with_alerts(provider: str, run_id: str, trace_id: str):
    def _do():
        data = fetch_from_provider(provider)  # your existing call
        if not data:
            raise ProviderEmptyDataError(f"Empty data from {provider}")
        return process_data(data)

    try:
        return run_with_retries(_do, _is_retryable, attempts=3)
    except ProviderEmptyDataError as e:
        ctx = {
            "stage": "ingest", "severity": "WARN", "provider": provider, 
            "run_id": run_id, "trace_id": trace_id, "error": str(e), "attempts": 3
        }
        emailer.send("provider_empty_data", f"[INGEST] Empty data: {provider}", ctx)
        store.put_alert(
            provider=provider, stage="ingest", severity="WARN", 
            code="PROVIDER_EMPTY", message="Empty data after retries", 
            metadata=ctx, dedupe_key=f"empty#{provider}#{run_id}"
        )
        raise
    except Exception as e:
        ctx = {
            "stage": "ingest", "severity": "ERROR", "provider": provider,
            "run_id": run_id, "trace_id": trace_id, "error": str(e), "attempts": 3
        }
        emailer.send("process_failure", f"[INGEST] Failure: {provider}", ctx)
        store.put_alert(
            provider=provider, stage="ingest", severity="ERROR",
            code="INGEST_FAILURE", message="Ingest failure after retries",
            metadata=ctx, dedupe_key=f"ingestfail#{provider}#{run_id}"
        )
        raise

Harmonize Worker Integration

from datetime import datetime, timezone, timedelta
from mesonet_alerts.emailer import EmailAlerter
from mesonet_alerts.store import AlertStore  
from mesonet_alerts.dropcheck import check_and_alert_volume_drop

emailer = EmailAlerter()
store = AlertStore()

# After harmonization run completes
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
window_end = now
window_start = now - timedelta(hours=1)

actual_count = count_harmonized_records(provider, window_start, window_end)  # your logic
check_and_alert_volume_drop(
    provider=provider,
    stage="harmonize", 
    actual_count=actual_count,
    expected_count=None,  # use ENV default for now
    threshold=0.20,
    window_start=window_start,
    window_end=window_end,
    emailer=emailer,
    store=store
)

🧪 Local Development Testing

Prerequisites

1. Build the wheel:

   cd micro-services/common/mesonet_alerts
   python -m build  # or uv build
   

2. Install into workers (already done in pyproject.toml):

   dependencies = [
       "mesonet-alerts @ file://../common/mesonet_alerts/dist/mesonet_alerts-0.1.0-py3-none-any.whl"
   ]
   

Usage (Local Dev)

Terminal 1 - Start Debug SMTP Server:

export $(grep -v '^#' .env.dev | xargs)
./scripts/run_debug_smtp.sh

Terminal 2 - Test Ingest Alerts:

export $(grep -v '^#' .env.dev | xargs)
cd micro-services/mesonet_ingest_worker
python scripts/test_ingest_alerts.py

Terminal 3 - Test Harmonize Alerts:

export $(grep -v '^#' .env.dev | xargs)
cd micro-services/mesonet_harmonize-worker
python scripts/test_harmonize_alerts.py

Expected Output:

• Terminal 1 should print full HTML+text email bodies
• Terminal 2/3 should show "✅ Alert sent" messages
• You should see nicely formatted emails with inline CSS

Environment Variables for Testing

# Override test parameters
TEST_PROVIDER=colorado        # Provider name for tests
TEST_ACTUAL=70               # Actual record count (harmonize test)
TEST_EXPECTED=100            # Expected record count (harmonize test)

🗄️ DynamoDB Schema

If ALERTS_TABLE_NAME is set, alerts are persisted with this schema:

Table: alerts
PK: alert_pk (String) = "{provider}#{stage}" 
SK: timestamp (String, ISO8601)
Attributes:
  - severity (String): ERROR, WARN, INFO
  - code (String): PROVIDER_EMPTY, INGEST_FAILURE, etc.
  - message (String): Human-readable message
  - metadata (Map): Additional context data
  - status (String): OPEN (default)
  - ttl (Number): Unix timestamp for auto-deletion
  - dedupe_key (String): Optional deduplication key
  - provider (String): Provider name
  - stage (String): Processing stage

🧪 Development

Running Tests

# Install dev dependencies
uv sync --dev

# Run tests
pytest tests/ -v

# With coverage
pytest tests/ --cov=src --cov-report=html

Local SMTP Testing

# Start local SMTP server for testing
python -m smtpd -c DebuggingServer -n localhost:1025

# Or use MailHog (recommended)
docker run -p 1025:1025 -p 8025:8025 mailhog/mailhog
# View emails at http://localhost:8025

🔮 Future Enhancements

The package includes commented hooks for future features:

Database-Backed Configuration

# TODO: Implement in config.py
EmailConfigRepo.get_active_config()  # SMTP from DB
RecipientRoutingRepo.get_recipients(provider, severity)  # Smart routing

Template Overrides

# TODO: Implement in templates.py
TemplateRepo.get(template_name, format_type)  # Custom templates from DB

EventBridge/SNS Integration

# TODO: Implement in store.py  
AlertEventPublisher.publish_alert_event(alert_data)  # Fan-out to external systems

Provider-Specific Volume Expectations

# TODO: Implement in dropcheck.py
VolumeExpectationRepo.get_expected_volume(provider, stage, hours)  # Smart baselines

📋 API Reference

EmailAlerter

• __init__(config=None, recipients=None) - Initialize with optional config override
• send(template, subject, context, recipients=None) - Send alert email
• resolve_recipients(provider, severity) - Future: smart recipient routing

AlertStore

• __init__(table_name=None) - Initialize with optional table name
• put_alert(provider, stage, severity, code, message, metadata=None, dedupe_key=None, ttl_seconds=86400) - Store alert
• get_recent_alerts(provider, stage, hours=24) - Retrieve recent alerts

Retry Functions

• run_with_retries(fn, is_retryable, attempts=3, backoffs=[1,3,9]) - Execute with retry logic
• retry_on_exceptions(*exception_types) - Decorator for auto-retry
• is_network_error(e), is_rate_limit_error(e), is_provider_error(e) - Error classifiers

Volume Drop Detection

• check_and_alert_volume_drop(**kwargs) - Check and alert on volume drops
• get_volume_trend(provider, stage, hours_back=24, store=None) - Analyze volume trends (placeholder)

🔒 Security Notes

• Credentials are read from environment variables only
• SMTP passwords are not logged
• DynamoDB uses IAM roles for authentication
• All database operations use conditional writes for consistency
• TTL automatically expires old alerts

📄 License

MIT License - see LICENSE file for details.