Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion src/videoipath_automation_tool/apps/inventory/app/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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.

Expand Down
19 changes: 17 additions & 2 deletions src/videoipath_automation_tool/apps/inventory/inventory_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down