[uniconfig] Update frinx-uniconfig-api to v1.0.0 and refactored uniconfig workers

uniconfig/python/CHANGELOG.md

@@ -12,3 +12,7 @@
 
 # 1.0.4
 - Enhanced 'handle_response' function for better error insights
+
+# 1.0.5
+- Updated API dependency frinx-uniconfig-api to rev "[frinx-uniconfig-api_v1.0.0](https://github.com/FRINXio/frinx-services-python-api/blob/main/uniconfig/python/CHANGELOG.md)"
+- Simplified response handling from UniConfig and refactored DeviceDiscovery

uniconfig/python/frinx_worker/uniconfig/__init__.py

@@ -1,13 +1,12 @@
 import dataclasses
+import http.client
 import json
 from typing import Any
 from typing import Optional
 
 from frinx.common.conductor_enums import TaskResultStatus
 from frinx.common.type_aliases import DictAny
 from frinx.common.util import remove_empty_elements_from_dict
-from frinx.common.worker.task_result import TO
-from frinx.common.worker.task_result import TaskResult
 from pydantic import BaseModel
 from requests import Response
 
@@ -17,37 +16,27 @@ class TransactionContext(BaseModel):
     transaction_id: str
 
 
-def handle_response(response: Response, worker_output: Optional[TO] = None) -> TaskResult:
-    common_log_info = (
-        f"URL: {response.url}; "
-        f"HTTP Request Status: {response.status_code} - {response.reason}; "
-        f"Method: {response.request.method}; "
-        f"Response content: '{response.content.decode('utf-8', errors='replace')[:200]}' "
-    )
-
-    def failed_task_result(reason: str) -> TaskResult:
-        return TaskResult(
-            status=TaskResultStatus.FAILED,
-            logs=f'{reason}; {common_log_info}'
-        )
-
-    if not response.ok:
-        return failed_task_result(f'HTTP request failed with status code {response.status_code}')
-
-    try:
-        if worker_output is not None:
-            worker_output.output = response.json()
-            output_status = worker_output.output.get('output', {}).get('status')
-            if output_status in ['fail','error']:
-                return failed_task_result('The response indicates failure')
+class UniconfigResultDetails(BaseModel):
+    logs: str
+    output: DictAny = {}
+    task_status: TaskResultStatus
 
-    except json.JSONDecodeError:
-        return failed_task_result('JSON decoding failed - unparsable response content')
 
-    return TaskResult(
-        status=TaskResultStatus.COMPLETED,
-        output=worker_output
+def handle_response(response: Response, model: Optional[type[BaseModel]] = None) -> UniconfigResultDetails:
+    uniconfig_result = UniconfigResultDetails(
+        task_status=TaskResultStatus.COMPLETED if response.ok else TaskResultStatus.FAILED,
+        logs=f'{response.request.method} request to {response.url} returned with status code {response.status_code}.'
     )
+    if response.status_code != http.client.NO_CONTENT:
+        try:
+            response_json = response.json()
+            if model is not None:
+                uniconfig_result.output = model.parse_obj(response_json).dict()
+            else:
+                uniconfig_result.output = response_json
+        except json.JSONDecodeError:
+            uniconfig_result.logs += 'ERROR: JSON decoding failed - unparsable response content.'
+    return uniconfig_result
 
 
 def uniconfig_zone_to_cookie(

uniconfig/python/frinx_worker/uniconfig/cli_network_topology.py

@@ -68,7 +68,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS,
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class Execute(WorkerImpl):
 
@@ -119,4 +126,11 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))

uniconfig/python/frinx_worker/uniconfig/connection_manager.py

@@ -63,7 +63,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class UninstallNode(WorkerImpl):
         from frinx_api.uniconfig.connection.manager import ConnectionType
@@ -106,7 +113,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class InstallMultipleNodes(WorkerImpl):
         from frinx_api.uniconfig.connection.manager.installmultiplenodes import Input
@@ -149,7 +163,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 headers=dict(UNICONFIG_HEADERS, accept='application/json')
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class UninstallMultipleNodes(WorkerImpl):
         from frinx_api.uniconfig.connection.manager.uninstallmultiplenodes import Input
@@ -193,4 +214,11 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))

uniconfig/python/frinx_worker/uniconfig/device_discovery.py

@@ -3,8 +3,10 @@
 
 import pydantic
 import requests
-from frinx.common.conductor_enums import TaskResultStatus
+from frinx.common.frinx_rest import UNICONFIG_HEADERS
+from frinx.common.frinx_rest import UNICONFIG_REQUEST_PARAMS
 from frinx.common.frinx_rest import UNICONFIG_URL_BASE
+from frinx.common.type_aliases import DictAny
 from frinx.common.type_aliases import ListStr
 from frinx.common.worker.service import ServiceWorkersImpl
 from frinx.common.worker.service import WorkerImpl
@@ -24,6 +26,7 @@
 from pydantic import IPvAnyNetwork
 
 from . import class_to_json
+from . import handle_response
 from .util import get_list_of_ip_addresses
 from .util import parse_ranges
 
@@ -70,15 +73,21 @@ def validate_ip(cls, ip: str) -> list[Address]:
                 return [address]
 
             @pydantic.field_validator('tcp_port', mode='before')
-            def validate_tcp(cls, tcp_port: str) -> list[TcpPortItem]:
-                return [TcpPortItem(port=p) for p in parse_ranges(tcp_port.split(','))]
+            def validate_tcp(cls, tcp_port: str) -> list[TcpPortItem] | None:
+                if tcp_port:
+                    return [TcpPortItem(port=p) for p in parse_ranges(tcp_port.split(','))]
+                else:
+                    return None
 
             @pydantic.field_validator('udp_port', mode='before')
-            def validate_udp(cls, udp_port: str) -> list[UdpPortItem]:
-                return [UdpPortItem(port=p) for p in parse_ranges(udp_port.split(','))]
+            def validate_udp(cls, udp_port: str) -> list[UdpPortItem] | None:
+                if udp_port:
+                    return [UdpPortItem(port=p) for p in parse_ranges(udp_port.split(','))]
+                else:
+                    return None
 
         class WorkerOutput(TaskOutput):
-            output: OperationsDiscoverPostResponse
+            output: DictAny
 
         def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
             if Discover.request is None:
@@ -95,6 +104,8 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
 
             response = requests.request(
                 url=UNICONFIG_URL_BASE + Discover.uri,
+                headers=dict(UNICONFIG_HEADERS),
+                params=UNICONFIG_REQUEST_PARAMS,
                 method=Discover.method,
                 data=class_to_json(
                     Discover.request(
@@ -103,22 +114,11 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 ),
             )
 
-            if response.ok:
-                status = TaskResultStatus.COMPLETED
-
-                return TaskResult(
-                    status=status,
-                    output=self.WorkerOutput(
-                        output=Discover.response(
-                            output=response.json()['output']
-                        )
-                    )
-                )
+            uniconfig_result = handle_response(response, OperationsDiscoverPostResponse)
 
-            status = TaskResultStatus.FAILED
             return TaskResult(
-                status=status,
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
                 output=self.WorkerOutput(
-                    output=response.json()
-                    )
-                )
+                    output=uniconfig_result.output,
+                ))

uniconfig/python/frinx_worker/uniconfig/snapshot_manager.py

@@ -65,7 +65,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class DeleteSnapshot(WorkerImpl):
         from frinx_api.uniconfig.rest_api import DeleteSnapshot as UniconfigApi
@@ -110,7 +117,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class ReplaceConfigWithSnapshot(WorkerImpl):
         from frinx_api.uniconfig.rest_api import ReplaceConfigWithSnapshot as UniconfigApi
@@ -158,4 +172,11 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))

uniconfig/python/frinx_worker/uniconfig/structured_data.py

@@ -66,7 +66,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(response, self.WorkerOutput(output=dict()))
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class WriteStructuredData(WorkerImpl):
         from frinx_api.uniconfig.rest_api import ReadStructuredData as UniconfigApi
@@ -123,14 +130,14 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(
-                response,
-                self.WorkerOutput(
-                    output=dict(
-                        response_code=response.status_code
-                    )
-                )
-            )
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))
 
     class DeleteStructuredData(WorkerImpl):
         from frinx_api.uniconfig.rest_api import DeleteStructuredData as UniconfigApi
@@ -187,11 +194,11 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 params=UNICONFIG_REQUEST_PARAMS
             )
 
-            return handle_response(
-                response,
-                self.WorkerOutput(
-                    output=dict(
-                        response_code=response.status_code
-                    )
-                )
-            )
+            uniconfig_result = handle_response(response)
+
+            return TaskResult(
+                status=uniconfig_result.task_status,
+                logs=uniconfig_result.logs,
+                output=self.WorkerOutput(
+                    output=uniconfig_result.output,
+                ))