Merge pull request #20 from dariomi/master

Add subscription functions

Author: andreasgraber

CHANGELOG.md

@@ -3,3 +3,4 @@
 > * v1.00, 11.2020 -- Adjustments for OpenSource - Andreas Graber
 > * v1.3, 06.2021 -- Snapshot creation added - Dario Kaelin
 > * v1.5, 01.2023 -- Dependency Update
+> * v1.6, 06.2024 -- Improvement for subscription - Dario Kaelin

README.md

@@ -102,6 +102,17 @@ You can specify a tenant in variable ```target_dn``` or not provide any to do a
 aci.snapshot(description='test', target_dn='/uni/tn-test')
 ```
 
+### Subscriptions
+You can subscribe to an ACI object with websocket and get near-instant updates on-change.  
+To use the subscriptions you have to:
+- Login to ACI
+- Open websocket to ACI
+- Subscribe to an ACI object via aciClient.subscribe
+- Refresh subscription to an ACI object via aciClient.subscription_refresh
+- Handle messages sent from ACI through websocket 
+
+You can find example code here: examples/subscription.py
+
 ## Testing
 
 ```

aciClient/aci.py

@@ -252,3 +252,56 @@ def snapshot(self, description="snapshot", target_dn="") -> bool:
         else:
             self.__logger.error(f'snapshot creation not succesfull: {response}')
             return False
+
+# ==============================================================================
+    # subscribe
+    # ==============================================================================
+    def subscribe(
+        self, subscription_dn: str, timeout: int = 60, query_parameters: list = []
+    ) -> {}:
+        query_parameters.append("subscription=yes")
+        query_parameters.append(f"refresh-timeout={timeout}")
+
+        endpoint = f"{self.baseUrl}{str(subscription_dn)}?{'&'.join(query_parameters)}"
+        self.__logger.debug(f"Subscribe to: {endpoint}")
+
+        response = self.session.get(endpoint, verify=False)
+        if response.status_code == 200:
+            self.__logger.debug(f"Successful subscribed to APIC: {response.json()}")
+            return response.json()
+        elif response.status_code == 400:
+            resp_text = (
+                f"400: {response.json()['imdata'][0]['error']['attributes']['text']}"
+            )
+            self.__logger.error(f"Error 400 during get occured: {resp_text}")
+            return response.json()
+        else:
+            self.__logger.error(f"Error during get occured: {response.json()}")
+            response.raise_for_status()
+            return response.json()
+
+    # ==============================================================================
+    # subscription_refresh
+    # ==============================================================================
+    def subscription_refresh(self, subscription_id: str) -> {}:
+        query_parameters = [f"id={subscription_id}"]
+
+        endpoint = (
+            f"{self.baseUrl}/subscriptionRefresh.json?{'&'.join(query_parameters)}"
+        )
+        self.__logger.debug(f"Refresh subscription: {subscription_id}")
+
+        response = self.session.post(endpoint, verify=False)
+        if response.status_code == 200:
+            self.__logger.debug(f"Successful subscribed to APIC: {response.json()}")
+            return response.json()
+        elif response.status_code == 400:
+            resp_text = (
+                f"400: {response.json()['imdata'][0]['error']['attributes']['text']}"
+            )
+            self.__logger.error(f"Error 400 during get occured: {resp_text}")
+            return response.json()
+        else:
+            self.__logger.error(f"Error during get occured: {response.json()}")
+            response.raise_for_status()
+            return response.json()

examples/subscription.py

@@ -0,0 +1,68 @@
+import json
+import time
+import ssl
+from threading import Thread
+
+from aciClient import ACI
+import websocket
+
+
+def on_message(ws, message):
+    print("Websocket received a message")
+    # Message contains the data transmited from ACI in plaintext
+    print(json.loads(message)["imdata"])
+
+
+def on_close(ws, close_status_code, close_msg):
+    print("Websocket was closed")
+
+
+def on_open(ws):
+    print("Websocket was opened")
+
+
+def open_websocket(ip: str, token: str):
+    ws = websocket.WebSocketApp(
+        f"wss://{ip}/socket{token}",
+        on_message=on_message,
+        on_open=on_open,
+        on_close=on_close,
+    )
+    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
+
+
+def main():
+    hostname = "sandboxapicdc.cisco.com"
+    username = "admin"
+    password = "!v3G@!4@Y"
+
+    aci = ACI(hostname, username, password)
+
+    # Login to ACI
+    aci.login()
+
+    # Open websocket to ACI with the login token
+
+    Thread(target=open_websocket, args=(aci.apicIp, aci.token)).start()
+    time.sleep(5)
+
+    # Subscribe to an ACI object
+    subscription_dn = "node/class/faultInst.json"
+    timeout = 60  # Optional. Default value in ACI
+    query_parameters = ["order-by=faultInst.lastTransition|asc"]  # Optional
+
+    response = aci.subscribe(
+        subscription_dn=subscription_dn,
+        timeout=timeout,
+        query_parameters=query_parameters,
+    )
+    subscription_id = response["subscriptionId"]
+
+    while True:
+        time.sleep(30)
+        print("Refreshing subscription")
+        aci.subscription_refresh(subscription_id=subscription_id)
+
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -7,10 +7,10 @@
     long_description = f.read()
 
 setup(name='aciClient',
-      version='1.5',
+      version='1.6',
       description='aci communication helper class',
       url='http://www.netcloud.ch',
-      author='mze',
+      author='Netcloud AG',
       author_email='nc_dev@netcloud.ch',
       license='MIT',
       packages=['aciClient'],