Generating Tokens using Python

Once you have set up a suitable Python environment, like that covered in the previous tutorial, it should now be straight-forward to automate many administrative tasks. This tutorial shows you how to generate bearer tokens automatically, so that additional (manual) steps are not needed when running a script to list all of your devices, for example.

Additionally, we will build up a Python class, DXAPI, that can be extended to develop scripts for many ThingPark Wireless activities, and we use a version of this code ourselves.

First Steps


import requests
import copy
from enum import Enum

class TokenPeriod(Enum):
    """
    Enumerated list of values for the 'validityPeriod' field of a token-generation request.
    """
    FiveMinutes = '5min'
    TwelveHours = '12hours'
    SevenDays = '7days'
    NinetyDays = '90days'
    Infinite = 'infinite'


class DXAPI:

    HEADER = {
        'Accept': 'application/json',
        'Connection': 'keep-alive',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/66.0.3359.181 Safari/537.36'
    }

    def __init__(self, url, api_profile, debug=False):
        """

        :param url: Base TPW URL, ie dx-api-au1.thingpark.com
        :param api_profile: Customer profile, ie iotnz-api
        :param debug:
        """
        self.baseurl = 'https://{}'.format(url)
        self.profile = api_profile
        self.debug = debug

    def __get(self, path, params=None, token=None):
        """

        :param path:
        :param params:
        :param token:
        :return:
        """
        url = '{}{}'.format(self.baseurl, path)
        header = copy.deepcopy(DXAPI.HEADER)
        if token:
            header['Authorization'] = 'Bearer {}'.format(token)
        resp = requests.get(url, params=params, headers=header)
        if resp.status_code != 200:
            msg = 'ERROR: {} - {}'.format(path, resp.status_code)
            raise RuntimeError(msg)
        return resp.json()

    def __post(self, path, params=None, token=None):
        """

        :param path:
        :param params:
        :param token:
        :return:
        """
        url = '{}{}'.format(self.baseurl, path)
        header = copy.deepcopy(DXAPI.HEADER)
        if token:
            header['Authorization'] = 'Bearer {}'.format(token)
        resp = requests.post(url, data=params, headers=header)
        if resp.status_code != 200:
            msg = 'ERROR: {} - {}'.format(path, resp.status_code)
            raise RuntimeError(msg)
        return resp.json()

    def get_token(self, username, password, period, renew=False):
        """
        Generate a bearer token for TPW
        :param username:
        :param password:
        :param period:
        :param renew:
        :return:
        """
        if not isinstance(period, TokenPeriod):
            raise Exception("token period is not a TokenPeriod object")

        params = {
            'grant_type': 'client_credentials',
            'renewToken': 'true' if renew else 'false',
            'validityPeriod': period.value,
            'client_id': '{}/{}'.format(self.profile, username),
            'client_secret': password
        }

        token = self.__post('/admin/latest/api/oauth/token', params)
        return token['access_token'], token

    def get_subscribers(self, token):
        """
        Get list of subscribers
        :param token:
        :return:
        """
        return self.__get('/core/latest/api/subscribers', token=token)

    def get_vendors(self, token):
        """
        Get list of vendors
        :param token:
        :return:
        """
        return self.__get('/core/latest/api/vendors', token=token)

Previous Steps


Next Steps