Notification handling

import datetime
from network_as_code import NetworkAsCodeApi

client = NetworkAsCodeApi(
    rapidapi_host="network-as-code.nokia.rapidapi.com",
    api_key="YOUR_API_KEY",
)

my_subscription = client.device_status.create_roaming_subscription(
    protocol="HTTP",
    sink="https://example.com/notifications",
    types=["org.camaraproject.device-roaming-status-subscriptions.v0.roaming-status"],
    config={
        "subscription_detail": {
            "device": {
                "phone_number": "+999991234567"
            }
        },
        "subscription_max_events": 5,
        "initial_event": True,
    },
    sink_credential={
        "access_token": "some-access-token",
        "access_token_expires_utc": (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=1)).isoformat(),
        "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 List, Optional, Union


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)