Example of How to Use the Onboarding API – Export Inventory to CSV

The User Experience Insight (UXI) Onboarding API allows you to simplify most of the tasks associated with assigning your sensors and agents to the proper groups and configuring them to test your networks and services. To use the API, your dashboard must be on the Greenlake Cloud Platform. Additionally, you must be using the group-based assignment feature.

You can find documentation for the Onboarding API here:

https://api.capenetworks.com/docs/openapi.yaml.

Download the OpenApi specification

Prerequisites

You will need to create a personal API client for the User Experience Insight service manager on the Greenlake Cloud Platform. See the detailed instructions on creating your personal API client in the Greenlake developer guide.

Example Scenario

In this example, suppose we want to get a complete inventory of all our resources and write the data to a collection of CSV files. This can be useful for inventory management, reporting and obtaining resource ID's for when we need to make additional calls to the onboarding API.

Endpoints

Here are the Endpoints you will need:

/networking-uxi/v1alpha1/groups
/networking-uxi/v1alpha1/sensors
/networking-uxi/v1alpha1/agents
/networking-uxi/v1alpha1/wireless-networks
/networking-uxi/v1alpha1/wired-networks
/networking-uxi/v1alpha1/service-tests
/networking-uxi/v1alpha1/sensor-group-assignments
/networking-uxi/v1alpha1/agent-group-assignments
/networking-uxi/v1alpha1/network-group-assignments
/networking-uxi/v1alpha1/service-test-group-assignments

Note: The API uses cursor-based pagination.

Putting it together

Here is a Python 3 example of this in action. This example uses Python requests to fetch the data and write it to a CSV file.

This example does the following:

Generates a temporary access token using the credentials of the personal API client.
Retrieves a list of all groups, sensors, agents, wired and wireless networks, service-tests as well as resource group assignments.
Writes the details of each resource to a CSV file specific to that resource type.

WARNING: This example is for demonstration purposes only. This example lacks robust error handling and optimizations necessary for a production environment. Do not use the example code in a production environment.

import csv
from typing import Any

import requests
from oauthlib.oauth2 import BackendApplicationClient
from requests.auth import HTTPBasicAuth
from requests_oauthlib import OAuth2Session

# Global constants
AUTH_TOKEN_URL = "https://sso.common.cloud.hpe.com/as/token.oauth2"
UXI_API_BASE_URL = "https://api.capenetworks.com/networking-uxi/v1alpha1"

# Customer specific constants
CLIENT_ID = "<CLIENT_ID>"
CLIENT_SECRET = "<CLIENT_SECRET>"

def main() -> None:
    api_token = get_api_token()

    groups = get_groups(api_token=api_token)
    write_to_csv("groups.csv", groups)

    sensors = get_sensors(api_token=api_token)
    write_to_csv("sensors.csv", sensors)

    agents = get_agents(api_token=api_token)
    write_to_csv("agents.csv", agents)

    wireless_networks = get_wireless_networks(api_token=api_token)
    write_to_csv("wireless_networks.csv", wireless_networks)

    wired_networks = get_wired_networks(api_token=api_token)
    write_to_csv("wired_networks.csv", wired_networks)

    service_tests = get_service_tests(api_token=api_token)
    write_to_csv("service_tests.csv", service_tests)

    sensor_group_assignments = get_sensor_group_assignments(api_token=api_token)
    write_to_csv("sensor_group_assignments.csv", sensor_group_assignments)

    agent_group_assignments = get_agent_group_assignments(api_token=api_token)
    write_to_csv("agent_group_assignments.csv", agent_group_assignments)

    network_group_assignments = get_network_group_assignments(api_token=api_token)
    write_to_csv("network_group_assignments.csv", network_group_assignments)

    service_test_group_assignments = get_service_test_group_assignments(api_token=api_token)
    write_to_csv("service_test_group_assignments.csv", service_test_group_assignments)


def get_api_token() -> str:
    client = BackendApplicationClient(CLIENT_ID)
    oauth = OAuth2Session(client=client)
    auth = HTTPBasicAuth(CLIENT_ID, CLIENT_SECRET)
    token = oauth.fetch_token(token_url=AUTH_TOKEN_URL, auth=auth)

    return token["access_token"]

def get_groups(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="groups", api_token=api_token)

def get_sensors(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="sensors", api_token=api_token)

def get_agents(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="agents", api_token=api_token)

def get_wireless_networks(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="wireless-networks", api_token=api_token)

def get_wired_networks(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="wired-networks", api_token=api_token)

def get_service_tests(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="service-tests", api_token=api_token)

def get_network_group_assignments(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="network-group-assignments", api_token=api_token)

def get_service_test_group_assignments(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="service-test-group-assignments", api_token=api_token)

def get_agent_group_assignments(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="agent-group-assignments", api_token=api_token)

def get_sensor_group_assignments(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="sensor-group-assignments", api_token=api_token)

def _get_items(path: str, api_token: str) -> list[dict[str, Any]]:
    HEADERS = {
        "Authorization": f"Bearer {api_token}",
    }
    items: list[dict[str, Any]] = []
    next_cursor = None

    # This while loop exists to keep making requests until all data for the specified resource is fetched.
    while True:
        try:
            params = {"next": next_cursor} if next_cursor else {}
            response = requests.get(
                f"{UXI_API_BASE_URL}/{path}", headers=HEADERS, params=params
            )
            response.raise_for_status()
            response_body: dict[str, Any] = response.json()
            response_items: list[dict[str, Any]] = response_body.get("items", [])
            next_cursor = response_body.get("next")

            items.extend(response_items)

            if not next_cursor or not response_items:
                break
        except requests.exceptions.HTTPError as e:
            print(f"Error fetching items ({path}): {e}")
            break

    return items

def write_to_csv(name: str, inventory: list[dict[str, Any]]) -> None:
    if not inventory:
        return

    for item in inventory:
        del item["type"]

    headers = list(inventory[0].keys())
    with open(name, "w", newline="") as file:
        writer = csv.writer(file)
        writer.writerow(headers)

        for item in inventory:
            writer.writerow(list(item.values()))

if __name__ == "__main__":
    main()