Module for accessing IMS API

inventory_provider/db/ims.py

+import requests
+import time
+
+from enum import Enum
+
+
+# Navigation Properties
+
+# http://149.210.162.190:81/ImsVersions/4.19.9/html/2d27d509-77cb-537d-3ffa-796de7e82af8.htm  # noqa
+NODE_PROPERTIES = {
+    'equipment definition': 16,
+    'management system': 128,
+    'site': 256,
+    'node city': 32768,
+    'order': 67108864,
+    'node country': 1073741824
+}
+# http://149.210.162.190:81/ImsVersions/4.19.9/html/347cb410-8c05-47bd-ceb0-d1dd05bf98a4.htm  # noqa
+CITY_PROPERTIES = {
+    'country': 8
+}
+# http://149.210.162.190:81/ImsVersions/4.19.9/html/347cb410-8c05-47bd-ceb0-d1dd05bf98a4.htm  # noqa
+EQUIP_DEF_PROPERTIES = {
+    'nodes': 4096
+}
+# http://149.210.162.190:81/ImsVersions/4.19.9/html/9c8d50f0-842c-5959-0fa6-14e1720669ec.htm  # noqa
+SITE_PROPERTIES = {
+    'aliases': 64,
+    'city': 2,
+    'country': 256
+}
+
+
+class InventoryStatus(Enum):
+    PLANNED = 1
+    READY_FOR_SERVICE = 2
+    IN_SERVICE = 3
+    MIGRATION = 4
+    OUT_OF_SERVICE = 6
+    READY_FOR_CEASURE = 7
+
+
+class IMSError(Exception):
+    pass
+
+
+class IMS(object):
+
+    TIMEOUT_THRESHOLD = 1200
+    NO_OF_ELEMENTS_PER_ITERATION = 50
+    PERMITTED_RECONNECT_ATTEMPTS = 3
+    LOGIN_PATH = '/login'
+    IMS_PATH = '/ims'
+
+    cache = {}
+    base_url = None
+    bearer_token = None
+    bearer_token_init_time = None
+    reconnect_attempts = 0
+
+    def __init__(self, base_url, username, password):
+        IMS.base_url = base_url
+        self.username = username
+        self.password = password
+
+    @classmethod
+    def _init_bearer_token(cls, username, password):
+        re_init_time = time.time()
+        if not cls.bearer_token or \
+                re_init_time - cls.bearer_token_init_time \
+                > cls.TIMEOUT_THRESHOLD:
+            IMS.reconnect_attempts = 0
+        else:
+            cls.reconnect_attempts += 1
+            if cls.reconnect_attempts > cls.PERMITTED_RECONNECT_ATTEMPTS:
+                raise IMSError('Too many reconnection attempts made')
+
+        response = requests.post(
+            IMS.base_url + IMS.LOGIN_PATH,
+            auth=(username, password))
+        response.raise_for_status()
+        cls.bearer_token_init_time = re_init_time
+        cls.bearer_token = response.json()
+
+    def _get_entity(
+            self,
+            url,
+            params=None,
+            navigation_properties=None,
+            use_cache=False):
+        url = f'{self.base_url + IMS.IMS_PATH}/{url}'
+        cache_key = url
+        if navigation_properties:
+            params = params if params else {}
+            params['navigationproperty'] = navigation_properties if isinstance(
+                navigation_properties, int) else sum(navigation_properties)
+
+        if use_cache:
+            if params:
+                s = '-'.join(
+                    [f'{t}::{params[t]}'
+                     for t in sorted(params)])
+                cache_key = f'{cache_key}::{s}'
+            entity = IMS.cache.get(cache_key, None)
+            if entity:
+                print('Got from cache')  # Remove after demo
+                return entity
+
+        if not IMS.bearer_token:
+            self._init_bearer_token(self.username, self.password)
+
+        make_request = True
+        while make_request:
+            response = requests.get(
+                url,
+                headers={'Authorization': self.bearer_token},
+                params=params)
+            if response.status_code == requests.codes.unauthorized:
+                IMS._init_bearer_token(self.username, self.password)
+            else:
+                make_request = False
+
+        response.raise_for_status()
+        return_value = response.json()
+
+        if use_cache:
+            IMS.cache[cache_key] = return_value
+        return return_value
+
+    def get_entity_by_id(
+            self,
+            entity_type,
+            entity_id,
+            navigation_properties=None,
+            use_cache=False):
+
+        url = f'{entity_type}/{entity_id}'
+        return \
+            self._get_entity(url, None, navigation_properties, use_cache)
+
+    def get_entity_by_name(
+            self,
+            entity,
+            name,
+            navigation_properties=None,
+            use_cache=False):
+        url = f'{entity}/byname/{name}'
+        return self._get_entity(url, None, navigation_properties, use_cache)
+
+    def get_filtered_entities(
+            self,
+            entity,
+            filter_string,
+            navigation_properties=None,
+            use_cache=False):
+        more_to_come = True
+        start_entity = 0
+        while more_to_come:
+            params = {
+                'paginatorStartElement': start_entity,
+                'paginatorNumberOfElements': IMS.NO_OF_ELEMENTS_PER_ITERATION
+            }
+            url = f'{entity}/filtered/{filter_string}'
+            entities = self._get_entity(
+                url,
+                params,
+                navigation_properties,
+                use_cache)
+            more_to_come = False
+            if entities:
+                more_to_come = len(entities) >= IMS.NO_OF_ELEMENTS_PER_ITERATION
+                start_entity += IMS.NO_OF_ELEMENTS_PER_ITERATION
+                yield from entities
+
+    def get_all_entities(
+            self,
+            entity,
+            navigation_properties=None,
+            use_cache=False):
+        yield from self.get_filtered_entities(
+            entity,
+            'Id <> 0',
+            navigation_properties,
+            use_cache)
+
+
+if __name__ == '__main__':
+    import sys
+    username = sys.argv[1]
+    password = sys.argv[2]
+    ims = IMS('http://83.97.94.128:81/api', username, password)
+    i = ims.get_entity_by_name(
+        'Site',
+        'London',
+        SITE_PROPERTIES.values())
+    print(i)
+
+