Python class for managing client system asset tracking with Device 42

d42lib.py

import getpass
import requests
import datetime

from requests.packages.urllib3.exceptions import SubjectAltNameWarning
requests.packages.urllib3.disable_warnings(SubjectAltNameWarning)


class D42Server(object):
    def __init__(self, url, username, password=None):
        if not password:
            password = getpass.getpass('Enter Device42 password: ')
        self.rooturl = url.rstrip('/')
        self.api = self.rooturl + '/api/1.0'
        self._session = requests.session()
        self._session.auth = (username, password)
        self._session.headers = {
            'Content-Type': 'application/x-www-form-urlencoded'}
        # test server connection
        self.test_connection()
        
    def test_connection(self):
        url = '{}/service_level/'.format(self.api)
        result = self._session.get(url)
        result.raise_for_status()

    def get_device(self, serial):
        url = '{}/devices/serial/{}/'.format(self.api, serial)
        result = self._session.get(url)
        data = result.json()
        if data.get('msg', 'OK') == 'device not found':
            return None
        return data

    def get_device_urls(self, device_name):
        url = '{}/device/url/'.format(self.api)
        params = { 'device': device_name }
        result = self._session.get(url, params=params)
        result.raise_for_status()
        return result.json()['device_weblinks']
        
    def add_device_url(self, device_name, url_type, **kwargs):
        url = '{}/device/url/'.format(self.api)
        params = { 'device': device_name, 'type': url_type }
        keys = ('host', 'port', 'url_suffix', 'notes')
        for key in set(kwargs.iterkeys()).intersection(keys):
            params[key] = kwargs[key]
        result = self._session.post(url, data=params)
        #result.raise_for_status()
        return result.json()

    def get_endusers(self):
        url = '{}/endusers/'.format(self.api)
        result = self._session.get(url)
        result.raise_for_status()
        return result.json()['values']

    def get_buildings(self):
        url = '{}/buildings/'.format(self.api)
        result = self._session.get(url)
        result.raise_for_status()
        return result.json()['buildings']

    def get_rooms(self, building_id=None):
        url = '{}/rooms/'.format(self.api)
        if building_id:
            params = { 'building_id': building_id }
            result = self._session.get(url, params=params)
        else:
            result = self._session.get(url)
        result.raise_for_status()
        return result.json()['rooms']

    def update_device(self, device_id, **kwargs):
        url = '{}/devices/'.format(self.api)
        params = { 'device_id': device_id }
        keys = ('asset_no', 'building', 'storage_room', 'storage_room_id', 
            'location', 'notes')
        for key in set(kwargs.iterkeys()).intersection(keys):
            params[key] = kwargs[key]
        result = self._session.put(url, data=params)
        #result.raise_for_status()
        return result.json()

    def add_lifecycle_event(self, event):
        url = '{}/lifecycle_event/'.format(self.api)
        params = {
            'date': event.date,
            'type': event.type,
            'device_id': str(event.device_id)
        }
        if event.user:
            params['user'] = event.user
        if event.notes:
            params['notes'] = event.notes
        result = self._session.put(url, data=params)
        #result.raise_for_status()
        return result.json()


class D42LifecycleEvent(object):
    def __init__(self, device_id, type_name, user=None, notes=None):
        self.date = datetime.datetime.now().strftime('%Y-%m-%d %H:%M')
        self.type = type_name
        self.device_id = device_id
        self.user = user
        self.notes = notes