[uniconfig] Add find-schema-resources-RPC, query-config RPC and ReadStructureData with jsonb filter

uniconfig/python/CHANGELOG.md

@@ -67,3 +67,6 @@
 
 # 3.0.2
 - Fix missing SNMP support in connection-manager UniConfig workers.
+
+# 3.1.0
+- Add find-schema-resources RPC

uniconfig/python/frinx_worker/uniconfig/schema_resources.py

@@ -0,0 +1,65 @@
+import requests
+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.worker.service import ServiceWorkersImpl
+from frinx.common.worker.task_def import TaskDefinition
+from frinx.common.worker.task_def import TaskExecutionProperties
+from frinx.common.worker.task_def import TaskInput
+from frinx.common.worker.task_def import TaskOutput
+from frinx.common.worker.task_result import TaskResult
+from frinx.common.worker.worker import WorkerImpl
+from frinx_api.uniconfig import OperationsFindSchemaResourcesPostResponse
+
+from . import class_to_json
+from . import handle_response
+from . import uniconfig_zone_to_cookie
+
+
+class SchemaResourcesWorkers(ServiceWorkersImpl):
+    class FindSchemaResources(WorkerImpl):
+        from frinx_api.uniconfig.rest_api import FindSchemaResources as UniconfigApi
+        from frinx_api.uniconfig.schema.resources.findschemaresources import Input
+
+        class ExecutionProperties(TaskExecutionProperties):
+            exclude_empty_inputs: bool = True
+            transform_string_to_json_valid: bool = True
+
+        class WorkerDefinition(TaskDefinition):
+            name: str = "UNICONFIG_Find_schema_resources_RPC"
+            description: str = "Find schema resources"
+
+        class WorkerInput(TaskInput):
+            repository_name: str
+            keywords: list[str]
+            transaction_id: str | None = None
+            uniconfig_server_id: str | None = None
+            uniconfig_url_base: str = UNICONFIG_URL_BASE
+
+        class WorkerOutput(TaskOutput):
+            output: OperationsFindSchemaResourcesPostResponse
+
+        def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
+            if self.UniconfigApi.request is None:
+                raise Exception(f"Failed to create request {self.UniconfigApi.request}")
+
+            response = requests.request(
+                url=worker_input.uniconfig_url_base + self.UniconfigApi.uri,
+                method=self.UniconfigApi.method,
+                data=class_to_json(
+                    self.UniconfigApi.request(
+                        input=self.Input(
+                            repository_name=worker_input.repository_name,
+                            keywords=worker_input.keywords,
+                        )
+                    )
+                ),
+                cookies=uniconfig_zone_to_cookie(
+                    uniconfig_server_id=worker_input.uniconfig_server_id, transaction_id=worker_input.transaction_id
+                ),
+                headers=dict(UNICONFIG_HEADERS),
+                params=UNICONFIG_REQUEST_PARAMS,
+            )
+
+            return handle_response(response, self.WorkerOutput)
+

uniconfig/python/frinx_worker/uniconfig/structured_data.py

@@ -67,6 +67,65 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
 
             return handle_response(response, self.WorkerOutput)
 
+    class ReadStructuredDataWithJsonBFilter(WorkerImpl):
+        from frinx_api.uniconfig.rest_api import ReadStructuredData as UniconfigApi
+
+        class ExecutionProperties(TaskExecutionProperties):
+            exclude_empty_inputs: bool = True
+
+        class WorkerDefinition(TaskDefinition):
+            name: str = "UNICONFIG_read_structured_device_data_with_jsonb_filter"
+            description: str = (
+                "Read device configuration or operational data in structured format e.g. openconfig "
+                "and filter the output using jsonb (JsonPath) filter"
+            )
+            labels: list[str] = ["BASICS", "UNICONFIG", "OPENCONFIG"]
+
+        class WorkerInput(TaskInput):
+            # url
+            node_id: str
+            uri: str | None = None
+            topology_id: str = "uniconfig"
+            # jsonb request parameters
+            jsonb_filter: str
+            add_parent_structure: bool = True
+            # uniconfig
+            transaction_id: str | None = None
+            uniconfig_server_id: str | None = None
+            uniconfig_url_base: str = UNICONFIG_URL_BASE
+
+        class WorkerOutput(TaskOutput):
+            output: DictAny
+
+        def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
+            uri = ""
+            if worker_input.uri:
+                if not worker_input.uri.startswith("/"):
+                    uri = f"/{worker_input.uri}"
+                else:
+                    uri = worker_input.uri
+
+            escaped_node_id = escape_uniconfig_uri_key(worker_input.node_id)
+            url = worker_input.uniconfig_url_base + self.UniconfigApi.uri.format(
+                topology_id=worker_input.topology_id, node_id=escaped_node_id, uri=uri
+            )
+
+            response = requests.request(
+                url=url,
+                method="POST",
+                cookies=uniconfig_zone_to_cookie(
+                    uniconfig_server_id=worker_input.uniconfig_server_id, transaction_id=worker_input.transaction_id
+                ),
+                headers={"Content-Type": "application/filter-data+json"},
+                params=UNICONFIG_REQUEST_PARAMS,
+                json={
+                    "query": worker_input.jsonb_filter,
+                    "addParentStructure": worker_input.add_parent_structure
+                }
+            )
+
+            return handle_response(response, self.WorkerOutput)
+
     class WriteStructuredData(WorkerImpl):
         from frinx_api.uniconfig.rest_api import ReadStructuredData as UniconfigApi
 

uniconfig/python/frinx_worker/uniconfig/uniconfig_query.py

@@ -0,0 +1,65 @@
+import requests
+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.worker.service import ServiceWorkersImpl
+from frinx.common.worker.task_def import TaskDefinition
+from frinx.common.worker.task_def import TaskExecutionProperties
+from frinx.common.worker.task_def import TaskInput
+from frinx.common.worker.task_def import TaskOutput
+from frinx.common.worker.task_result import TaskResult
+from frinx.common.worker.worker import WorkerImpl
+
+from . import class_to_json
+from . import handle_response
+from . import uniconfig_zone_to_cookie
+
+
+class UniconfigQueryWorkers(ServiceWorkersImpl):
+    class UniconfigQuery(WorkerImpl):
+        from frinx_api.uniconfig.rest_api import QueryConfig as UniconfigApi
+        from frinx_api.uniconfig.uniconfig.query.queryconfig import Input as UniconfigQueryConfigInput
+
+
+        class ExecutionProperties(TaskExecutionProperties):
+            exclude_empty_inputs: bool = True
+            transform_string_to_json_valid: bool = True
+
+        class WorkerDefinition(TaskDefinition):
+            name: str = "UNICONFIG_Query_config_RPC"
+            description: str = "Application JSONB filtering (JSONPath with dot/bracket notation)"
+
+        class WorkerInput(TaskInput, UniconfigQueryConfigInput):
+            transaction_id: str | None = None
+            uniconfig_server_id: str | None = None
+            uniconfig_url_base: str = UNICONFIG_URL_BASE
+
+        class WorkerOutput(TaskOutput):
+            output: DictAny
+
+        def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
+            if self.UniconfigApi.request is None:
+                raise Exception(f"Failed to create request {self.UniconfigApi.request}")
+
+            response = requests.request(
+                url=worker_input.uniconfig_url_base + self.UniconfigApi.uri,
+                method=self.UniconfigApi.method,
+                data=class_to_json(
+                    self.UniconfigApi.request(
+                        input=self.UniconfigQueryConfigInput(
+                            jsonb_path_query=worker_input.jsonb_path_query,
+                            node_id=worker_input.node_id,
+                            topology_id=worker_input.topology_id
+                        )
+                    )
+                ),
+                cookies=uniconfig_zone_to_cookie(
+                    uniconfig_server_id=worker_input.uniconfig_server_id, transaction_id=worker_input.transaction_id
+                ),
+                headers=dict(UNICONFIG_HEADERS),
+                params=UNICONFIG_REQUEST_PARAMS,
+            )
+
+            return handle_response(response, self.WorkerOutput)
+