Notification handling

import network_as_code as nac
from network_as_code.models.device import DeviceIpv4Addr
from network_as_code.models.device_status import EventType, AccessTokenCredential

client = nac.NetworkAsCodeClient(
    token="<your-application-key-here>"
)

my_device = client.devices.get(
    # Provide either the public and private address
    # or the public address and port.
    # Devices cannot be identified by their IPv4 address alone.
    ipv4_address=DeviceIpv4Addr(
        public_address="233.252.0.2",
        private_address="192.0.2.25",
        public_port=80
    ),
    ipv6_address="2001:db8:1234:5678:9abc:def0:fedc:ba98",
    # The phone number does not accept spaces or parentheses
    phone_number="+36721601234567"
)

# Here, we will keep track of a devices roaming status
# with a limit of 5 reports
# Simply change the event_type whenever needed
my_subscription = client.device_status.subscribe(
    device=my_device,
    event_type=[EventType.ROAMING_STATUS],
    # Use HTTPS to send notifications
    sink="https://example.com/notify",
    max_num_of_reports=5,
    sink_credential=AccessTokenCredential(
        access_token= "some-access-token",
        access_token_expires_utc= datetime.now(timezone.utc) + timedelta(days=1),
        access_token_type="bearer"
    )
)

# status_handler.py

# run with: uvicorn status_handler:app

from fastapi import FastAPI, Header
from pydantic import BaseModel

from typing_extensions import Annotated
from typing import Optional


app = FastAPI()

class Device(BaseModel):
    phoneNumber: Optional[str] | None
    networkAccessIdentifier: Optional[str] | None
    ipv4Address: Optional[str] | None
    ipv6Address: Optional[str] | None

class RoamingEventDetail(BaseModel):
    device: Device
    subscriptionId: str
    roaming: bool | None
    countryCode: int | None
    countryName: List[str] | None
    lastStatusTime: str | None
    terminationReason: str

class Event(BaseModel):
    eventType: str
    eventTime: str
    eventDetail: RoamingEventDetail

class Data(BaseModel):
    device: Device
    subscriptionId: str
    terminationReason: str

class Notification(BaseModel):
    id: str
    source: str
    type: str
    specversion: str
    datacontenttype: str
    time: str
    eventSubscriptionId: str
    event: Event
    data: Data


@app.post("/notifications")
def receive_notification(
    notification: Notification,
    authorization: Annotated[Union[str, None], Header]
):
    if authorization == "Bearer my-token":
        # We can now react to the notifications
        # based on the Notification object
        print(notification)