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()

Congestion notifications parameters

Here is a parameters table describing each parameter used in the code:

ParametersDescription
deviceInformation about the device for which congestion data is requested or subscribed to.
idIt represents the subscription identifier.
subscription_expire_timeExpiration time for the congestion subscription (using ISO 8601 format)
notification_urlURL where notifications about congestion updates will be sent.
notification_auth_token (Optional)Authentication token for accessing the notification URL.
maxNumberOfReports (Optional)Specify an integer value for the maximum number of reports to be sent in the subscription.
startStart timestamp for retrieving congestion data. It should be an ISO 8601 formatted date string or a date-time object as shown above.
endEnd timestamp for retrieving congestion data. It should be an ISO 8601 formatted date string or a date-time object as shown above.

Last updated on May 23, 2024