From 61a63cf2bb2e1982dc4b6f25e91fb92c80233ff9 Mon Sep 17 00:00:00 2001 From: Paul Winterstein Date: Mon, 19 May 2025 11:00:45 +0200 Subject: [PATCH] feat: Add method to list device IDs by driver --- .../apps/inventory/app/app.py | 12 +++++++++++- .../apps/inventory/inventory_api.py | 19 +++++++++++++++++-- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/src/videoipath_automation_tool/apps/inventory/app/app.py b/src/videoipath_automation_tool/apps/inventory/app/app.py index 121a4ad..e9e8cb5 100644 --- a/src/videoipath_automation_tool/apps/inventory/app/app.py +++ b/src/videoipath_automation_tool/apps/inventory/app/app.py @@ -9,7 +9,7 @@ ) from videoipath_automation_tool.apps.inventory.app.get_device import InventoryGetDeviceMixin from videoipath_automation_tool.apps.inventory.inventory_api import InventoryAPI -from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettings, CustomSettingsType +from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettings, CustomSettingsType, DriverLiteral from videoipath_automation_tool.apps.inventory.model.inventory_device import InventoryDevice from videoipath_automation_tool.apps.inventory.model.inventory_device_configuration_compare import ( InventoryDeviceComparison, @@ -229,6 +229,16 @@ def find_device_id_by_label( else: raise ValueError(f"Invalid label_search_mode: {label_search_mode}") + def list_device_ids_by_driver(self, driver: DriverLiteral) -> List[str]: + """Method to list all device ids by driver id. + + Args: + driver (DriverLiteral): Driver to filter devices by (e.g. `com.nevion.arista-0.1.0`). + Returns: + List[str]: List of device ids. + """ + return self._inventory_api.fetch_device_ids_by_driver(driver=driver) + def enable_device(self, device_id: str) -> InventoryDevice[CustomSettings]: """Method to enable a device in VideoIPath-Inventory. diff --git a/src/videoipath_automation_tool/apps/inventory/inventory_api.py b/src/videoipath_automation_tool/apps/inventory/inventory_api.py index 194d698..e6c32f4 100644 --- a/src/videoipath_automation_tool/apps/inventory/inventory_api.py +++ b/src/videoipath_automation_tool/apps/inventory/inventory_api.py @@ -7,9 +7,12 @@ from pydantic import IPvAnyAddress from typing_extensions import deprecated -from videoipath_automation_tool.apps.inventory.inventory_utils import construct_driver_id_from_info +from videoipath_automation_tool.apps.inventory.inventory_utils import ( + construct_driver_id_from_info, + extract_driver_info_from_id, +) from videoipath_automation_tool.apps.inventory.model.device_status import DeviceStatus -from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettingsType +from videoipath_automation_tool.apps.inventory.model.drivers import CustomSettingsType, DriverLiteral from videoipath_automation_tool.apps.inventory.model.inventory_device import InventoryDevice from videoipath_automation_tool.apps.inventory.model.inventory_discovered_device import DiscoveredInventoryDevice from videoipath_automation_tool.apps.inventory.model.inventory_request_rpc import InventoryRequestRpc @@ -368,6 +371,18 @@ def fetch_device_ids_list(self) -> List[str]: raise ValueError("Response data is empty.") return [device["_id"] for device in response.data["config"]["devman"]["devices"]["_items"]] + def fetch_device_ids_by_driver(self, driver: DriverLiteral) -> List[str]: + """Method to fetch all device ids by driver id from VideoIPath-Inventory""" + driver_organization, driver_name, driver_version = extract_driver_info_from_id(driver_id=driver) + escaped_driver_organization = urllib.parse.quote(driver_organization, safe="") + escaped_driver_name = urllib.parse.quote(driver_name, safe="") + escaped_driver_version = urllib.parse.quote(driver_version, safe="") + url_path = f"/rest/v2/data/config/devman/devices/* where (config.driver.name='{escaped_driver_name}' and config.driver.version='{escaped_driver_version}' and config.driver.organization='{escaped_driver_organization}') /**" + response = self.vip_connector.rest.get(url_path) + if not response.data: + raise ValueError("Response data is empty.") + return [device["_id"] for device in response.data["config"]["devman"]["devices"]["_items"]] + # --- Bulk Device Label Fetching Methods --- def fetch_devices_factory_labels_as_dict(self) -> dict[str, str]: """Method to fetch all device factory labels from VideoIPath-Inventory