Congestion notifications

Subscribe or unsubscribe from Congestion notifications so you can stay up to date with different device congestion levels.

Subscribing to notifications via webhook

Here's how you can subscribe to Congestion level change notifications:

import network_as_code as nac
 
from datetime import datetime, timezone, timedelta
 
from network_as_code.models.device import Device, DeviceIpv4Addr
 
# Create a client and device objects as previously shown
client = nac.NetworkAsCodeClient(...)
 
my_device = client.devices.get(...)
 
# Subscribe your device to Congestion notifications
congestion_subscription = client.insights.subscribe_to_congestion_info(
    my_device,
    # Set the duration of your subscription to congestion insights,
    # e.g.: it can end in `n` days starting from now.
    subscription_expire_time=datetime.now(timezone.utc) + timedelta(days=1),
    # Set a notification URL with auth token to receive updates
    notification_url="https://example.com/notify",
    notification_auth_token="my-secret-token"
)
 
# Subscriptions are identified by id, for management
# Use this to show the subscription:
print(congestion_subscription.id)
 
# Or check when your subscription starts/expires:
print(my_subscription.starts_at)
print(my_subscription.expires_at)

With this snippet, you can subscribe to congestion updates using the subscribe_to_congestion_info method. It passes the device information, subscription expiration time, notification URL, and notification authentication token as parameters. The method returns a subscription object.

It then prints the ID of the subscription for management purposes.

Congestion level notification handler

# congestion_handler.py
 
# run with: uvicorn congestion_handler:app
 
from fastapi import FastAPI, Header
from pydantic import BaseModel
 
from typing_extensions import Annotated
from typing import Union
 
app = FastAPI()
 
class Data(BaseModel):
    level: str
 
class Notification(BaseModel):
    id: str
    source: str
    type: str
    specversion: str
    datacontenttype: str
    time: str
    data: Data
 
@app.post("/notifications")
def receive_notification(
    notification: Notification,
    authorization: Annotated[Union[str, None], Header]
):
    if authorization == "Bearer my-secret-token":
        # We can now react to the notifications
        # based on the Notification object
        print(notification)

Unsubscribing from congestion notifications

Finally, the code shows how to stop a subscription by calling the delete method. You simply call the delete() method with your congestion subscription object to delete it.

 
# Use the congestion_subscription object previously created
# and delete it to stop receiving notifications:
congestion_subscription.delete()