Fetching Ping Results from a Specific Agent using NetBeez API into a CSV file

APIs & Developer Support
1

In this guide, we’ll demonstrate how to write a Python script to fetch all ping results from a specific agent within a given time range, using NetBeez’s API. The results will be saved in a CSV file, with each timestamp parsed into a human-readable date/time.

Prerequisites

  • Python 3.x
  • A valid Bearer token for NetBeez’s API

Installation

First, install the requests library to handle HTTP requests:

pip install requests

Step 1: import the necessary libraries:

import requests
import csv
import time
from datetime import datetime

Step 2: define some constants:

URL = "https://[HOSTNAME]/nb_tests/ping/results"
BEARER_TOKEN = "YOUR_BEARER_TOKEN"
HEADERS = {'Authorization': f'Bearer {BEARER_TOKEN}'}
CSV_FILE = "ping_results.csv"
TIME_CHUNK = 60 * 1000 # One minute in milliseconds
agent_id = "226"
start_timestamp = round((time.time() - (60 * 60)) * 1000)
end_timestamp = round(time.time() * 1000)

Step 3: (optional) Disable insecure warnings:

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

Step 4: Define a function to fetch the results and write them into a CSV file.

The function takes the agent id and start and end times as input. It then loops through in 1 minutes chunks and retrieves all the data within each minute, while using a 100 page limit pagination loop.

def fetch_data(agent_id, start_timestamp, end_timestamp):
    with open('ping_results.csv', 'w', newline='') as csvfile:
        fieldnames = ['value', 'network_interface_type', 'error_code', 'ts', 'agent_id', 'test_template_id', 'wifi_profile']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        current_timestamp = start_timestamp
        while current_timestamp <= end_timestamp:
            params = {
                "filter[agents]": agent_id,
                "filter[ts][operator]": "<=>",
                "filter[ts][value1]": current_timestamp,
                "filter[ts][value2]": current_timestamp + 60000,
                "order[attributes]": 'ts',
                "order[direction]": 'asc',
                "page[limit]": 100,
                "page[offset]": 1,
            }
            
            next_page = True
            
            while next_page:
                try:
                    response = requests.get(URL, headers=HEADERS, params=params, verify=False)
                    response.raise_for_status()
                except requests.RequestException as e:
                    print(f"An error occurred: {e}")
                    continue
                
                data = response.json()
                for result in data['data']:
                    row = {
                        'value': result['attributes']['value'],
                        'network_interface_type': result['attributes']['network_interface_type'],
                        'error_code': result['attributes']['error_code'],
                        'ts': datetime.fromtimestamp(result['attributes']['ts'] / 1000).strftime('%Y-%m-%d %H:%M:%S'),
                        'agent_id': result['relationships']['agent']['data']['id'],
                        'test_template_id': result['relationships']['nb_test_template']['data']['id'],
                        'wifi_profile': result['relationships']['wifi_profile']['data']['id'] if result['relationships']['wifi_profile']['data'] else None,
                    }
                    writer.writerow(row)
                    print(f"{row}")
                
                next_page = data['meta']['page']['next']
                params["page[offset]"] += 1
            
            current_timestamp += TIME_CHUNK

Step 5: Run the function

fetch_data(agent_id, start_timestamp, end_timestamp)

Conclusion

This script fetches ping results from the NetBeez API in one-minute intervals and writes them to a CSV file. Make sure to replace HOSTNAME with your server’s fqdn, and YOUR_BEARER_TOKEN with your actual token. If you encounter any issues or have any questions/suggestions, feel free to reach out in the comments below!

Colab: If you would like to play with this on your own environment but don’t want to setup too many things then use this Google Colab notebook that I have prepared for this post: Google Colab

1 Like