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:
Parameters | Description |
---|---|
device | Information about the device for which congestion data is requested or subscribed to. |
id | It represents the subscription identifier. |
subscription_expire_time | Expiration time for the congestion subscription (using ISO 8601 format) |
notification_url | URL 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. |
start | Start timestamp for retrieving congestion data. It should be an ISO 8601 formatted date string or a date-time object as shown above. |
end | End 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 21, 2024