Consolidate NetBox device lookup into centralized utility

osism/api.py

@@ -22,6 +22,7 @@
 
 from osism.tasks import reconciler, openstack
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 from osism.services.listener import BaremetalEvents
 from osism.services.websocket_manager import websocket_manager
 from osism.services.event_bridge import event_bridge
@@ -159,31 +160,7 @@ class BaremetalNodesResponse(BaseModel):
     count: int = Field(..., description="Total number of nodes")
 
 
-def find_device_by_identifier(identifier: str):
-    """Find a device in NetBox by various identifiers."""
-    if not utils.nb:
-        return None
-
-    device = None
-
-    # Search by device name
-    devices = utils.nb.dcim.devices.filter(name=identifier)
-    if devices:
-        device = list(devices)[0]
-
-    # Search by inventory_hostname custom field
-    if not device:
-        devices = utils.nb.dcim.devices.filter(cf_inventory_hostname=identifier)
-        if devices:
-            device = list(devices)[0]
-
-    # Search by serial number
-    if not device:
-        devices = utils.nb.dcim.devices.filter(serial=identifier)
-        if devices:
-            device = list(devices)[0]
-
-    return device
+# find_device_by_identifier has been moved to osism.utils.netbox
 
 
 @app.get("/", tags=["health"])

osism/commands/baremetal.py

@@ -16,6 +16,7 @@
 
 from osism.commands import get_cloud_connection
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 from osism.tasks.conductor.netbox import get_nb_device_query_list_ironic
 from osism.tasks import netbox
 from osism.utils.ssh import cleanup_ssh_known_hosts_for_node
@@ -59,16 +60,8 @@ def take_action(self, parsed_args):
             device_role = "N/A"
             if utils.nb:
                 try:
-                    # Try to find device by name first
-                    device = utils.nb.dcim.devices.get(name=b["name"])
-
-                    # If not found by name, try by inventory_hostname custom field
-                    if not device:
-                        devices = utils.nb.dcim.devices.filter(
-                            cf_inventory_hostname=b["name"]
-                        )
-                        if devices:
-                            device = list(devices)[0]
+                    # Use centralized device lookup function
+                    device = find_device_by_identifier(b["name"])
 
                     # Get device role
                     if device and device.role and hasattr(device.role, "name"):
@@ -206,16 +199,8 @@ def take_action(self, parsed_args):
                 default_vars = {}
                 if utils.nb:
                     try:
-                        # Try to find device by name first
-                        device = utils.nb.dcim.devices.get(name=node.name)
-
-                        # If not found by name, try by inventory_hostname custom field
-                        if not device:
-                            devices = utils.nb.dcim.devices.filter(
-                                cf_inventory_hostname=node.name
-                            )
-                            if devices:
-                                device = devices[0]
+                        # Use centralized device lookup function
+                        device = find_device_by_identifier(node.name)
 
                         # Extract local_context_data if device found and has the field
                         if (
@@ -343,16 +328,8 @@ def take_action(self, parsed_args):
                 default_vars = {}
                 if utils.nb:
                     try:
-                        # Try to find device by name first
-                        device = utils.nb.dcim.devices.get(name=node.name)
-
-                        # If not found by name, try by inventory_hostname custom field
-                        if not device:
-                            devices = utils.nb.dcim.devices.filter(
-                                cf_inventory_hostname=node.name
-                            )
-                            if devices:
-                                device = devices[0]
+                        # Use centralized device lookup function
+                        device = find_device_by_identifier(node.name)
 
                         # Extract local_context_data if device found and has the field
                         if (
@@ -441,14 +418,8 @@ def take_action(self, parsed_args):
                     logger.error("NetBox connection not available")
                     return
 
-                # Try to find device by name first
-                device = utils.nb.dcim.devices.get(name=name)
-
-                # If not found by name, try by inventory_hostname custom field
-                if not device:
-                    devices = utils.nb.dcim.devices.filter(cf_inventory_hostname=name)
-                    if devices:
-                        device = devices[0]
+                # Use centralized device lookup function
+                device = find_device_by_identifier(name)
 
                 # If device not found, error out
                 if not device:
@@ -695,10 +666,11 @@ def take_action(self, parsed_args):
 
         try:
             if name:
-                devices = [utils.nb.dcim.devices.get(name=name)]
-                if not devices[0]:
+                device = find_device_by_identifier(name)
+                if not device:
                     logger.error(f"Device {name} not found in NetBox")
                     return
+                devices = [device]
             else:
                 # Use the NETBOX_FILTER_CONDUCTOR_IRONIC setting to get devices
                 devices = set()
@@ -1244,7 +1216,7 @@ def take_action(self, parsed_args):
                         f"Clearing NetBox states for node {node.name} on primary NetBox"
                     )
                     try:
-                        device = utils.nb.dcim.devices.get(name=node.name)
+                        device = find_device_by_identifier(node.name)
                         if device:
                             device.custom_fields.update(
                                 {"provision_state": None, "power_state": None}

osism/commands/console.py

@@ -9,6 +9,7 @@
 from prompt_toolkit import prompt
 
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 from osism.utils.ssh import ensure_known_hosts_file, KNOWN_HOSTS_PATH
 
 
@@ -46,7 +47,7 @@ def get_primary_ipv4_from_netbox(hostname: str) -> Optional[str]:
         return None
 
     try:
-        device = utils.nb.dcim.devices.get(name=hostname)
+        device = find_device_by_identifier(hostname)
         if device and device.primary_ip4:
             ip_address = str(device.primary_ip4.address).split("/")[0]
             logger.info(f"Found primary IPv4 for {hostname} in Netbox: {ip_address}")

osism/commands/netbox.py

@@ -10,6 +10,7 @@
 
 from osism.tasks import conductor, netbox, handle_task
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 
 
 class Ironic(Command):
@@ -300,21 +301,17 @@ def take_action(self, parsed_args):
             logger.error("NetBox integration not configured.")
             return
 
-        # Search for device by name first
-        devices = list(utils.nb.dcim.devices.filter(name=host))
-
-        # If not found by name, search by custom fields
-        if not devices:
-            # Search by alternative_name custom field
-            devices = list(utils.nb.dcim.devices.filter(cf_alternative_name=host))
-
-        if not devices:
-            # Search by inventory_hostname custom field
-            devices = list(utils.nb.dcim.devices.filter(cf_inventory_hostname=host))
-
-        if not devices:
-            # Search by external_hostname custom field
-            devices = list(utils.nb.dcim.devices.filter(cf_external_hostname=host))
+        # Use centralized device lookup with extended search fields
+        device = find_device_by_identifier(
+            host,
+            search_fields=[
+                "name",
+                "cf_alternative_name",
+                "cf_inventory_hostname",
+                "cf_external_hostname",
+            ],
+        )
+        devices = [device] if device else []
 
         if not devices:
             logger.error(f"Device '{host}' not found in NetBox.")

osism/commands/sonic.py

@@ -12,6 +12,7 @@
 from tabulate import tabulate
 
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 from osism.tasks import netbox
 from osism.tasks.conductor.netbox import (
     get_nb_device_query_list_sonic,
@@ -34,17 +35,12 @@ class SonicCommandBase(Command):
 
     def _get_device_from_netbox(self, hostname):
         """Get device from NetBox by name or inventory_hostname"""
-        device = utils.nb.dcim.devices.get(name=hostname)
+        device = find_device_by_identifier(hostname)
         if not device:
-            devices = utils.nb.dcim.devices.filter(cf_inventory_hostname=hostname)
-            if devices:
-                device = devices[0]
-                logger.info(f"Device found by inventory_hostname: {device.name}")
-            else:
-                logger.error(
-                    f"Device {hostname} not found in NetBox (searched by name and inventory_hostname)"
-                )
-                return None
+            logger.error(
+                f"Device {hostname} not found in NetBox (searched by name and inventory_hostname)"
+            )
+            return None
         return device
 
     def _get_config_context(self, device, hostname):

osism/tasks/conductor/sonic/sync.py

@@ -5,6 +5,7 @@
 from loguru import logger
 
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 from osism.tasks.conductor.netbox import get_nb_device_query_list_sonic
 from .bgp import calculate_minimum_as_for_group
 from .connections import (
@@ -61,7 +62,7 @@ def sync_sonic(device_name=None, task_id=None, show_diff=True):
     if device_name:
         # When specific device is requested, fetch it directly
         try:
-            device = utils.nb.dcim.devices.get(name=device_name)
+            device = find_device_by_identifier(device_name)
             if device:
                 # Check if device role matches allowed roles
                 if device.role and device.role.slug in DEFAULT_SONIC_ROLES:

osism/tasks/conductor/utils.py

@@ -5,6 +5,7 @@
 from loguru import logger
 
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 import sushy
 import urllib3
 
@@ -109,14 +110,8 @@ def get_redfish_connection(
     # Try to find NetBox device first for conductor configuration fallback
     if utils.nb:
         try:
-            # First try to find device by name
-            device = utils.nb.dcim.devices.get(name=hostname)
-
-            # If not found by name, try by inventory_hostname custom field
-            if not device:
-                devices = utils.nb.dcim.devices.filter(cf_inventory_hostname=hostname)
-                if devices:
-                    device = devices[0]
+            # Use centralized device lookup function
+            device = find_device_by_identifier(hostname)
         except Exception as exc:
             logger.warning(f"Could not resolve hostname {hostname} via NetBox: {exc}")
 

osism/utils/netbox.py

@@ -0,0 +1,64 @@
+# SPDX-License-Identifier: Apache-2.0
+
+from loguru import logger
+from osism import utils
+
+
+def find_device_by_identifier(identifier: str, search_fields: list = None):
+    """
+    Find a NetBox device by multiple identifier types.
+
+    Searches through various device fields in priority order to locate
+    a device in NetBox. Default search order: name, cf_inventory_hostname, serial.
+
+    Args:
+        identifier: Device identifier (name, hostname, serial number, etc.)
+        search_fields: List of field names to search.
+                      Default: ['name', 'cf_inventory_hostname', 'serial']
+
+    Returns:
+        Device object if found, None otherwise
+
+    Examples:
+        >>> device = find_device_by_identifier('server-01')
+        >>> device = find_device_by_identifier('host123', ['cf_inventory_hostname'])
+    """
+    if not utils.nb:
+        logger.debug("NetBox connection not available")
+        return None
+
+    if not identifier or not str(identifier).strip():
+        logger.debug("Empty identifier provided")
+        return None
+
+    identifier = str(identifier).strip()
+
+    if search_fields is None:
+        search_fields = ["name", "cf_inventory_hostname", "serial"]
+
+    for field in search_fields:
+        try:
+            logger.debug(f"Searching for device by {field}: {identifier}")
+
+            if field == "name":
+                # Use get() for name field (expects unique result)
+                device = utils.nb.dcim.devices.get(name=identifier)
+                if device:
+                    logger.debug(f"Found device '{device.name}' by {field}")
+                    return device
+            else:
+                # Use filter() for custom fields and serial
+                devices = utils.nb.dcim.devices.filter(**{field: identifier})
+                if devices:
+                    device = list(devices)[0]
+                    logger.debug(f"Found device '{device.name}' by {field}")
+                    return device
+
+        except (StopIteration, IndexError):
+            continue
+        except Exception as e:
+            logger.debug(f"Error searching by {field}: {e}")
+            continue
+
+    logger.warning(f"Device '{identifier}' not found in NetBox")
+    return None

osism/utils/ssh.py

@@ -6,6 +6,7 @@
 from loguru import logger
 
 from osism import utils
+from osism.utils.netbox import find_device_by_identifier
 
 
 # Default path for SSH known_hosts file
@@ -82,7 +83,7 @@ def get_host_identifiers(hostname: str) -> List[str]:
     # Try Netbox fallback if available
     if utils.nb:
         try:
-            device = utils.nb.dcim.devices.get(name=hostname)
+            device = find_device_by_identifier(hostname)
             if device and device.primary_ip4:
                 ip_address = str(device.primary_ip4.address).split("/")[0]
                 if ip_address and ip_address not in identifiers: