Merge pull request #209 from netdevopsbr/generic

Closes #208 - Refactor "netbox_proxbox.backend.routes.netbox.generic" subpackage and split it into smaller subpackages.

netbox_proxbox/backend/cache.py

@@ -5,9 +5,8 @@ def __init__(self):
         self.cache: dict = {}
 
     def get(self, key: str):
-        result = self.cache.get(key)
-        if result is not None:
-            return result
+        result = self.cache.get(key, {})
+        return result
 
     def set(self, key: str, value: Any):
         self.cache[key] = value

netbox_proxbox/backend/routes/netbox/generic/check_duplicate/__init__.py

@@ -0,0 +1,4 @@
+from .check_default import _check_default
+from .check_pk_address import _check_pk_address
+from .check_pk_virtual_machine import _check_pk_virtual_machine
+from .check_vm_interface import _check_vm_interface

netbox_proxbox/backend/routes/netbox/generic/check_duplicate/check_default.py

@@ -0,0 +1,69 @@
+from netbox_proxbox.backend.exception import ProxboxException
+from netbox_proxbox.backend.logging import log
+
+from fastapi import WebSocket
+
+import asyncio
+
+async def _check_default(
+    websocket: WebSocket,
+    pynetbox_path,
+    default_name: str,
+    default_slug: str,
+    object_name: str,
+    nb, 
+
+):
+    """
+    Asynchronously checks for default objects in Netbox before creating a new one.
+    """
+
+    await log(
+        websocket=websocket,
+        msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Checking default object."
+    )
+
+    try:
+        # Check if the default object exists in Netbox.
+        result = await asyncio.to_thread(
+            pynetbox_path.get,
+            name=default_name,
+            slug=default_slug,
+            tag=[nb.tag.slug]
+        )
+
+        # If the default object exists, return it.
+        if result:
+            return result
+
+        else:
+            # If no object found searching using tag, try to find without it, using just name and slug.
+            result = await asyncio.to_thread(
+                pynetbox_path.get,
+                name=default_name,
+                slug=default_slug,
+            )
+
+            # If the default object exists, return it.
+            if result:
+                await log(
+                    websocket=websocket,
+                    msg=f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Default <strong>{object_name}</strong> with ID <strong>{result.id}</strong> found on Netbox, but without Proxbox tag.\nPlease delete it (or add the tag) and try again.\nNetbox does not allow duplicated names and/or slugs."
+                )
+            else:
+                return None
+
+        # If the default object does not exist, return None.
+        return None
+
+    except ProxboxException as error:
+        await log(
+            websocket=websocket,
+            msg=f'{error}'
+        )
+
+    except Exception as error:
+        await log(
+            websocket=websocket,
+            msg=f"<span class='badge text-bg-red' title='Post'><strong><i class='mdi mdi-upload'></i></strong></span> Error trying to create default <strong>{object_name}</strong> on Netbox.\nPython Error: {error}",
+        )

netbox_proxbox/backend/routes/netbox/generic/check_duplicate/check_pk_address.py

@@ -0,0 +1,54 @@
+from netbox_proxbox.backend.logging import log
+
+from fastapi import WebSocket
+
+import asyncio
+
+async def _check_pk_address(
+    websocket: WebSocket,
+    pynetbox_path,
+    primary_field_value: str,
+    object_name: str,
+):
+    await log(
+        websocket=websocket,
+        msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Checking duplicate OBJECT using the ADDRESS as PRIMARY FIELD."
+    )
+
+    result_by_address = None
+
+    try:
+        result_by_address = await asyncio.to_thread(
+            pynetbox_path.get,
+            address=primary_field_value
+        )
+
+    except Exception as error:
+        if "get() returned more than one result" in f"{error}":
+            try:
+                result_by_filter_address = await asyncio.to_thread(pynetbox_path.filter, address=primary_field_value)
+
+                if result_by_filter_address:
+                    for address in result_by_filter_address:
+                        print(f"ADDRESS OBJECT: {address}")
+                        # TODO: Check if the address object is the same as the one being created.
+
+            except:
+                await log(
+                    websocket=websocket,
+                    msg=f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Error trying to <code>FILTER</code> <strong>{object_name}</strong> by address on Netbox.\nPython Error: {error}",
+                )
+
+
+        await log(
+            websocket=websocket,
+            msg=f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Error trying to get <strong>{object_name}</strong> by address on Netbox.\nPython Error: {error}",
+        )
+
+    if result_by_address:
+        await log(
+            websocket=websocket,
+            msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> IP Address with the same network found. Returning it."
+        )
+
+        return result_by_address

netbox_proxbox/backend/routes/netbox/generic/check_duplicate/check_pk_virtual_machine.py

@@ -0,0 +1,77 @@
+from netbox_proxbox.backend.logging import log
+
+from fastapi import WebSocket
+
+import asyncio
+
+async def _check_pk_virtual_machine(
+    websocket: WebSocket,
+    pynetbox_path,
+    primary_field_value: str,
+    object_name: str,
+    endpoint: str
+):
+    await log(
+        websocket=websocket,
+        msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Checking duplicate device using the VIRTUAL MACHINE as PRIMARY FIELD."
+    )
+
+    if endpoint == "interfaces":                  
+        result_by_virtual_machine = None
+
+        try:
+            # THE ERROR IS HERE.
+            #
+            # GET
+            result_by_virtual_machine = await asyncio.to_thread(
+                pynetbox_path.get,
+                virtual_machine=primary_field_value
+            )
+
+            if result_by_virtual_machine:
+                for interface in result_by_virtual_machine:
+                    print(f"INTERFACE OBJECT: {interface} | {interface.virtual_machine}")
+
+                    print(f"interface.virtual_machine: {interface.virtual_mchine} | primary_field_value: {primary_field_value}")
+
+                    # Check if Virtual Machine registered on the interface is the same as the one being created.
+                    if interface.virtual_machine == primary_field_value:
+                        return interface
+
+                    else:
+                        return None
+
+        except Exception as error:
+            await log(
+                websocket=websocket,
+                msg=f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Error trying to get interface using only 'virtual_machine' field as parameter.\n   >{error}"
+            )
+
+            if "get() returned more than one result" in f"{error}":
+                # FILTER
+                await log(
+                    websocket=websocket,
+                    msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> Found more than one <strong>VM INTERFACE</strong> object with the same <strong>virtual_machine</strong> field.\nTrying to use <code>.filter</code> pynetbox method now."
+                )
+
+                try:
+                    result_by_virtual_machine = await asyncio.to_thread(
+                        pynetbox_path.filter,
+                        virtual_machine=primary_field_value,
+                    )
+
+                    if result_by_virtual_machine:
+                        for interface in result_by_virtual_machine:
+                            print(f"INTERFACE OBJECT: {interface} | {interface.virtual_machine}")
+
+                            print(f"interface.virtual_machine: {interface.virtual_mchine} | primary_field_value: {primary_field_value}")
+                            if interface.virtual_machine == primary_field_value:
+                                return interface
+                            else:
+                                return None
+
+                except Exception as error:
+                    await log(
+                        websocket=websocket,
+                        msg=f"Error trying to get 'VM Interface' object using 'virtual_machine' and 'name' fields.\nPython Error: {error}",
+                    )

netbox_proxbox/backend/routes/netbox/generic/check_duplicate/check_vm_interface.py

@@ -0,0 +1,54 @@
+from netbox_proxbox.backend.logging import log
+
+from fastapi import WebSocket
+
+import asyncio
+
+async def _check_vm_interface(
+    websocket: WebSocket,
+    pynetbox_path,
+    primary_field_value: str,
+    vm_interface_object: dict
+):
+    result_by_vm_interface = None
+
+    try:
+        # GET
+        result_by_vm_interface = await asyncio.to_thread(
+            pynetbox_path.get,
+            virtual_machine=primary_field_value,
+            name=vm_interface_object.get("name", "")
+        )
+
+        await log(
+            websocket=websocket,
+            msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> If duplicate interface found, check if the virtual machines are the same."
+        )
+
+        # Check if result is not None.
+        if result_by_vm_interface:
+            await log(
+                websocket=websocket,
+                msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> VM Interface with the same Virtual Machine ID found. Duplicated object, returning it."
+            )
+
+            # Check if Virtual Machine registered on the VM Interface is the same as the one being created.
+            if result_by_vm_interface.virtual_machine == primary_field_value:
+                return result_by_vm_interface
+
+            # If the Virtual Machine is different, return as NOT duplicated. Interface NAME is the same, but the Virtual Machine fields are different.
+            else:
+                await log(
+                    websocket=websocket,
+                    msg="<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> If interface equal, but different devices, return as NOT duplicated."
+                )
+
+                return None
+
+    except Exception as error:
+        await log(
+            websocket=websocket,
+            msg=f"Error trying to get 'VM Interface' object using 'virtual_machine' and 'name' fields.\nPython Error: {error}",
+        )
+
+    return None