Notifications

The Doover notification system lets agents send alerts to users via email, SMS, web push, or HTTP webhooks. Notifications are published to an agent's notifications channel and fanned out to matching subscriptions by the Doover cloud. The DataClient provides methods for sending notifications, managing delivery endpoints, and configuring subscriptions.

Sending a Notification

The simplest way to send a notification is with send_notification, which publishes a message to the agent's notifications channel:

from pydoover.api import DataClient

client = DataClient(profile="default")

# Send a basic notification
message = client.send_notification(
    agent_id=12345,
    message="Temperature exceeded threshold: 92.3C",
    title="High Temperature Alert",
)

print(f"Notification sent as message {message.id}")

Severity Levels

Notifications have a severity level that determines which subscribers receive them. Subscribers only receive notifications at or above their subscription severity. The available severity levels are:

Severity	Value	Use Case
`Trace`	3	Verbose diagnostic information
`Debug`	4	Debugging details
`Info`	5	General informational messages
`Warn`	6	Warning conditions
`Critical`	7	Critical alerts requiring immediate attention

from pydoover.models.data.notification import NotificationSeverity

# Send a critical alert
client.send_notification(
    agent_id=12345,
    message="Sensor failure detected on input 3",
    title="Sensor Failure",
    severity=NotificationSeverity.Critical,
)

# Send an informational notification
client.send_notification(
    agent_id=12345,
    message="Daily report generated successfully",
    severity=NotificationSeverity.Info,
)

Topics

Topics provide an additional filtering dimension. Subscribers can set topic_filter to receive only notifications on specific topics:

# Send a notification with a topic
client.send_notification(
    agent_id=12345,
    message="Battery below 20%",
    title="Low Battery",
    severity=NotificationSeverity.Warn,
    topic="battery",
)

Using a Notification Object

For more control, you can construct a Notification object and pass it directly:

from pydoover.models.data.notification import Notification, NotificationSeverity

notification = Notification(
    message="Pump motor overheating",
    title="Motor Alert",
    severity=NotificationSeverity.Critical,
    topic="motors",
)

client.send_notification(agent_id=12345, message=notification)

Fetching Notifications

Retrieve the notification state for an agent:

# Get the agent's notification configuration and recent state
response = client.fetch_notifications(agent_id=12345)

Notification Endpoints

Endpoints define where notifications are delivered. Each endpoint has a type, a name, and type-specific configuration in extra_data.

Endpoint Types

Type	Value	`extra_data` Fields
`Email`	1	`{"email": "user@example.com"}`
`Sms`	2	`{"phone_number": "+61400000000"}`
`WebPush`	3	Web push subscription details
`Http`	4	`{"url": "https://webhook.example.com"}`

Listing Endpoints

# List all notification endpoints for an agent
endpoints = client.list_notification_endpoints(agent_id=12345)

for ep in endpoints:
    print(f"{ep.name} ({ep.type.name}): {ep.extra_data}")

Creating an Email Endpoint

from pydoover.models.data.notification import NotificationType

# Create an email notification endpoint
endpoint_id = client.create_notification_endpoint(
    agent_id=12345,
    name="Operations Team Email",
    type=NotificationType.Email,
    extra_data={"email": "ops-team@example.com"},
    default=True,
)

print(f"Created endpoint: {endpoint_id}")

Setting default=True means this endpoint will be used for new subscriptions that do not specify an explicit endpoint.

Updating an Endpoint

# Update the endpoint name and email address
client.update_notification_endpoint(
    agent_id=12345,
    endpoint_id=endpoint_id,
    name="Updated Ops Email",
    extra_data={"email": "ops-alerts@example.com"},
)

Testing an Endpoint

Send a test notification to verify an endpoint is configured correctly:

success = client.test_notification_endpoint(
    agent_id=12345,
    endpoint_id=endpoint_id,
)

if success:
    print("Test notification delivered successfully")

Deleting an Endpoint

client.delete_notification_endpoint(
    agent_id=12345,
    endpoint_id=endpoint_id,
)

Notification Subscriptions

Subscriptions connect a subscriber agent to a target agent's notifications, filtering by severity and topic. When an agent sends a notification, the cloud checks all subscriptions where subscribed_to matches that agent and the severity and topic filters are satisfied.

Creating a Subscription

from pydoover.models.data.notification import NotificationSeverity

# Subscribe agent 100 to receive notifications from agent 12345
# Only receive Warn-level and above, filtered to the "temperature" topic
subscriptions = client.create_notification_subscription(
    agent_id=100,
    subscribe_to=12345,
    severity=NotificationSeverity.Warn,
    topic_filter=["temperature"],
)

for sub in subscriptions:
    print(f"Subscription {sub['id']} on endpoint {sub['endpoint_id']}")

If you omit endpoint_id, the subscription is created for all of the agent's default endpoints.

Listing Subscriptions

# List all subscriptions for an agent
subs = client.list_notification_subscriptions(agent_id=100)

for sub in subs:
    print(f"Subscribed to agent {sub.subscribed_to}, severity >= {sub.severity.name}")

You can also filter to show only subscriptions targeting a specific agent:

# List subscriptions that target agent 12345
subs = client.list_notification_subscriptions(
    agent_id=100,
    subscribed_to=12345,
)

Updating a Subscription

# Change the severity filter
client.update_notification_subscription(
    agent_id=100,
    subscription_id=sub_id,
    severity=NotificationSeverity.Critical,
    topic_filter=["temperature", "pressure"],
)

Deleting a Subscription

client.delete_notification_subscription(
    agent_id=100,
    subscription_id=sub_id,
)

Default Subscriptions

Default subscriptions are organisation-level defaults that apply to new agents automatically. They are managed separately from per-agent subscriptions:

# List default subscriptions
defaults = client.list_default_notification_subscriptions(agent_id=100)

# Delete a default subscription
client.delete_default_notification_subscription(
    agent_id=100,
    subscribed_to=12345,
)

Listing Subscribers

To see which agents are subscribed to receive notifications from a given agent:

subscribers = client.list_notification_subscribers(agent_id=12345)

for sub in subscribers:
    print(f"Agent {sub.subscriber} subscribed at severity {sub.severity.name}")

Data Client -- DataClient constructor and configuration
Channels -- channel operations (notifications use the notifications channel)
Alarms -- alarm rules that can trigger notifications