Technical overview > API > Use case: inventory with REST API

In this use case, we learn how to trigger a device inventory request by the use of API. The motivational reasons are similar to the discovery use case.

Requirements

•Python 3+

•Requests library (https://docs.python-requests.org/en/latest/)

•API key

•frontend URL

•the new device is connected to the runner

•credentials with device access to the new device are registered

•site to relate the device to

Program content

import requests
import datetime
import time

APIKEY = "PUT API KEY HERE"
FRONTEND = "PUT FRONTEND URL HERE"

HEADERS = { "ApiKey": APIKEY }

def datestring(TP):
    return str(TP.year) + "-" + str(TP.month) + "-" + str(TP.day) + "T" + str(TP.hour) + ":" + str(TP.minute) + ":" + str(TP.second) + "Z"

DEVICE_HOSTNAME = "test-device"
DEVICE_CREATION_DATE = datestring(datetime.datetime.now())
DEVICE_SITE_ID = None

def obtain_paged(EP):
    P = {}
    P["count"] = 1000

    items = []

    while True:
        result = requests.get(EP, params=P, headers=HEADERS)

        if result.ok == False:
            break

        objs = result.json()

        if objs == None:
            return

        if len(objs) == 0:
            return

        items.extend(objs)

        lastobj = objs[len(objs)-1]

        P["LastId"] = lastobj["id"]

    return items

def describe_api_error(OPD, R):
    print("failed to " + OPD + " (" + str(R.status_code) + ", " + R.reason + ")" )

    try:
        error_details = R.json()

        if error_details != None and "details" in error_details:
            print( "error details: " + error_details["details"] )
    except:
        pass

def main():
    devs = obtain_paged(FRONTEND + "/api/v1/Devices")

    does_already_exist = False

    if devs != None:
        for D in devs:
            if "name" in D and D["name"] == DEVICE_HOSTNAME:
                does_already_exist = True
                break

    if does_already_exist:
        print( "device does already exist (" + FRONTEND + ", " + DEVICE_HOSTNAME + ")" )
        return

    sites = obtain_paged(FRONTEND + "/api/v1/Site")

    if sites == None or len(sites) == 0:
        print( "failed to obtain sites from frontend (" + FRONTEND + ")" )
        return

    siteid = DEVICE_SITE_ID

    for S in sites:
        if "name" in S and S["name"] == "Default Site":
            siteid = S["id"]
            break

    if siteid == None:
        siteid = sites[0]["id"]

    postdata = {
        "device": {
            "name": DEVICE_HOSTNAME,
            "creationDate": DEVICE_CREATION_DATE,
            "comment": "imported by API",
            "corporateOwnership": False
        },
        "metadata": {
            "siteId": siteid
        }
    }

    R = requests.post(FRONTEND + "/api/v1/Operations/device-import", headers=HEADERS, json=postdata)

    if R.ok == False:
        describe_api_error("import device", R)
        return

    print( "added device to Raynet One" )

    attempt_idx = 0

    devid = None

    while devid == None and attempt_idx < 3:
        devs = obtain_paged(FRONTEND + "/api/v1/Devices")

        if devs == None:
            print( "FATAL: failed to obtain devices after device import; remove the obsolete device manually" )
            return

        for D in devs:
            if "name" in D and D["name"] == DEVICE_HOSTNAME:
                devid = D["id"]
                break

        attempt_idx = attempt_idx + 1

    if devid == None:
        print( "failed to find device after device import" )
        return

    postdata = {
        "useCustomSettings": False,
        "targetingType": "device",
        "includeDevicesWithoutInventory": True,
        "deviceIds": [devid]
    }

    R = requests.post(FRONTEND + "/api/v1/Operations/inventory", headers=HEADERS, json=postdata)

    if R.ok:
        print( "triggered device inventory of " + DEVICE_HOSTNAME )
    else:
        describe_api_error("trigger device inventory", R)
        return