Start fixing Ruff corrections manually

netdevopsbr · Oct 28, 2024 · 2bd939a · 2bd939a
1 parent 51b0a3a
commit 2bd939a
Show file tree

Hide file tree

Showing 19 changed files with 162 additions and 194 deletions.
diff --git a/netbox_proxbox/backend/logging.py b/netbox_proxbox/backend/logging.py
@@ -1,6 +1,6 @@
 import logging
 from logging.handlers import TimedRotatingFileHandler
-from netbox_proxbox.backend.exception import ProxboxException
+from fastapi import WebSocket
 
 # ANSI escape sequences for colors
 class AnsiColorCodes:
@@ -73,19 +73,15 @@ def setup_logger():
 
 logger = setup_logger()
 
-from fastapi import WebSocket
-
 async def log(websocket: WebSocket, msg, level = None):
     if websocket: 
         await websocket.send_text(msg)
 
-    if level == "debug": logger.debug(msg)
+    if level == "debug":
+        logger.debug(msg)
 
     if level == "ERROR" or level == "error":
         logger.error(msg)
-        #raise ProxboxException(
-        #    message=msg,
-        #    python_exception=python_exception
-        #)
 
-    else: logger.info(msg)
+    else:
+        logger.info(msg)
diff --git a/netbox_proxbox/backend/routes/netbox/__init__.py b/netbox_proxbox/backend/routes/netbox/__init__.py
@@ -51,7 +51,7 @@ async def netbox_devices(nb: NetboxSessionDep):
     return raw_list
 
 @router.get("/openapi")
-async def netbox_devices(nb: NetboxSessionDep):
+async def netbox_openapi(nb: NetboxSessionDep):
     """
     ### Fetches the OpenAPI documentation from the Netbox session.
     

diff --git a/netbox_proxbox/backend/routes/netbox/dcim/__init__.py b/netbox_proxbox/backend/routes/netbox/dcim/__init__.py
@@ -8,11 +8,8 @@
 from .device_roles import DeviceRole
 from .manufacturers import Manufacturer
 
-from netbox_proxbox.backend.schemas.netbox import CreateDefaultBool
 from netbox_proxbox.backend.schemas.netbox.dcim import SitesSchema
 
-from netbox_proxbox.backend.logging import logger
-
 # FastAPI Router
 router = APIRouter()
 

diff --git a/netbox_proxbox/backend/routes/netbox/dcim/device_roles.py b/netbox_proxbox/backend/routes/netbox/dcim/device_roles.py
@@ -1,5 +1,4 @@
 from netbox_proxbox.backend.routes.netbox.generic import NetboxBase
-from typing import Any
 
 class DeviceRole(NetboxBase):
     """

diff --git a/netbox_proxbox/backend/routes/netbox/dcim/device_types.py b/netbox_proxbox/backend/routes/netbox/dcim/device_types.py
@@ -1,8 +1,6 @@
 from netbox_proxbox.backend.routes.netbox.generic import NetboxBase
 from .manufacturers import Manufacturer
 
-from typing import Any
-
 class DeviceType(NetboxBase):
 
     # Default Device Type Params

diff --git a/netbox_proxbox/backend/routes/netbox/dcim/devices.py b/netbox_proxbox/backend/routes/netbox/dcim/devices.py
@@ -1,5 +1,4 @@
 from netbox_proxbox.backend.routes.netbox.generic import NetboxBase
-from typing import Any
 
 from .sites import Site
 from .device_types import DeviceType
@@ -9,13 +8,13 @@
 
 class Device(NetboxBase):
 
-    default_name = "Proxbox Basic Device"
-    default_slug = "proxbox-basic-device"
-    default_description = "Proxbox Basic Device"
+    default_name: str = "Proxbox Basic Device"
+    default_slug: str = "proxbox-basic-device"
+    default_description: str = "Proxbox Basic Device"
 
-    app = "dcim"
-    endpoint = "devices"
-    object_name = "Device"
+    app: str = "dcim"
+    endpoint: str = "devices"
+    object_name: str = "Device"
 
     async def get_base_dict(self):
         site = await Site(nb = self.nb, websocket = self.websocket).get()

diff --git a/netbox_proxbox/backend/routes/netbox/dcim/interfaces.py b/netbox_proxbox/backend/routes/netbox/dcim/interfaces.py
@@ -1,22 +1,16 @@
 from netbox_proxbox.backend.routes.netbox.generic import NetboxBase
-from typing import Any
-
-from .sites import Site
-from .device_types import DeviceType
-from .device_roles import DeviceRole
 
 from netbox_proxbox.backend.routes.netbox.dcim.devices import Device
 
 class Interface(NetboxBase):
 
-    default_name = "Proxbox Basic interface"
-    #default_slug = "proxbox-basic-interface"
-    default_description = "Proxbox Basic Interface"
-    type = "other"
+    default_name: str = "Proxbox Basic interface"
+    default_description: str = "Proxbox Basic Interface"
+    type: str = "other"
 
-    app = "dcim"
-    endpoint = "interfaces"
-    object_name = "Interface"
+    app: str = "dcim"
+    endpoint: str = "interfaces"
+    object_name: str = "Interface"
 
     async def get_base_dict(self):
         device = await Device(nb = self.nb, websocket = self.websocket).get()

diff --git a/netbox_proxbox/backend/routes/netbox/generic.py b/netbox_proxbox/backend/routes/netbox/generic.py
@@ -1,4 +1,4 @@
-from fastapi import Query, Body
+from fastapi import Query
 
 from typing import Annotated
 from typing_extensions import Doc
@@ -104,20 +104,23 @@ async def get(
         print(kwargs)
         await log(self.websocket, f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> Getting <strong>{self.object_name}</strong> from Netbox.")
 
-        if self.id: return await self._get_by_id()
-        if self.all: return await self._get_all()
+        if self.id:
+            return await self._get_by_id()
 
-        if kwargs: return await self._get_by_kwargs(**kwargs)
+        if self.all:
+            return await self._get_all()
+
+        if kwargs:
+            return await self._get_by_kwargs(**kwargs)
 
-
         if self.pynetbox_path.count() == 0:
 
             await log(self.websocket, f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> There's no <strong>{self.object_name}</strong> registered on Netbox. Creating a DEFAULT ONE.")
 
-            self.default = True
+            self.default: bool = True
             create_default_object = await self.post()
 
-            if create_default_object != None:
+            if create_default_object is None:
                 await log(self.websocket, f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> Default <strong>{self.object_name}</strong> created successfully. {self.object_name} ID: {create_default_object.id}")
                 return create_default_object
 
@@ -152,17 +155,23 @@ async def get(
                     }
                 )
 
-            if get_object != None:
+            if get_object is None:
                 await log(self.websocket, f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> The <strong>{self.object_name}</strong> found is from 'Proxbox' (because it has the tag). Returning it.")
                 return get_object
 
             # 2.2.2. If it's not Proxbox one, create a default one.
-            await log(self.websocket, f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> The <strong>{self.object_name}</strong> object found IS NOT from 'Proxbox'. Creating a default one.")
-            self.default = True
+            await log(
+                self.websocket,
+                f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> The <strong>{self.object_name}</strong> object found IS NOT from 'Proxbox'. Creating a default one."
+            )
+
+            self.default: bool = True
+
             default_object = await self.post()
             return default_object
 
-        except ProxboxException as error: raise error
+        except ProxboxException as error:
+            raise error
 
         except Exception as error:
             raise ProxboxException(
@@ -210,9 +219,9 @@ async def _get_by_kwargs(self, **kwargs):
                         else:
                             await log(self.websocket, "<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 ProxboxException as error: raise error
+   
+        except ProxboxException as error:
+            raise error
 
         except Exception as error:
             raise ProxboxException(
@@ -253,19 +262,20 @@ async def _get_by_id(self):
                 )
 
             # 1.1. Return found object.
-            if response != None:
+            if response is None:
                 await log(self.websocket, f"<span class='badge text-bg-blue' title='Get'><strong><i class='mdi mdi-download'></i></strong></span> <strong>{self.object_name}</strong> with ID <strong>{self.id}</strong> found on Netbox. Returning it.")
                 return response
 
             # 1.2. Raise ProxboxException if object is not found.
             else:
                 raise ProxboxException(
                     message=f"<span class='text-blue'><strong><i class='mdi mdi-download'></i></strong></span> <span class='text-grey'><strong>[GET]</strong></span> <strong>{self.object_name}</strong> with ID <strong>{self.id}</strong> not found on Netbox.",
-                    detail=f"Please check if the ID provided is correct. If it is, please check if the object has the Proxbox tag. (You can use the 'ignore_tag' query parameter to ignore this check and return object without Proxbox tag)"
+                    detail="Please check if the ID provided is correct. If it is, please check if the object has the Proxbox tag. (You can use the 'ignore_tag' query parameter to ignore this check and return object without Proxbox tag)"
                 )
 
 
-        except ProxboxException as error: raise error
+        except ProxboxException as error:
+            raise error
 
         except Exception as error:
             raise ProxboxException(
@@ -358,18 +368,19 @@ async def post(
         if data:
             await log(self.websocket, f"<span class='badge text-bg-red' title='Post'><strong><i class='mdi mdi-upload'></i></strong></span> Creating <strong>{self.object_name}</strong> object on Netbox.")
 
-            if isinstance(data, dict) == False:
+            if isinstance(data, dict) is False:
                 try:
                     # Convert Pydantic model to Dict through 'model_dump' Pydantic method.
                     data = data.model_dump(exclude_unset=True)
+
                 except Exception as error:
                     raise ProxboxException(
-                        message=f"<span class='text-red'><strong><i class='mdi mdi-upload'></i></strong></span> <span class='text-red'><strong><i class='mdi mdi-error'></i></strong></span> <strong>[POST]</strong> Error parsing Pydantic model to Dict.",
+                        message="<span class='text-red'><strong><i class='mdi mdi-upload'></i></strong></span> <span class='text-red'><strong><i class='mdi mdi-error'></i></strong></span> <strong>[POST]</strong> Error parsing Pydantic model to Dict.",
                         python_exception=f"{error}",
                     )
 
             # If no explicit slug was provided by the payload, create one based on the name.
-            if data.get("slug") == None:
+            if data.get("slug") is None:
                 if not self.primary_field:
                     await log(self.websocket, "<span class='badge text-bg-red' title='Post'><strong><i class='mdi mdi-upload'></i></strong></span> <strong>SLUG</strong> field not provided on the payload. Creating one based on the NAME or MODEL field.")
                     try:
@@ -380,27 +391,27 @@ async def post(
                             data["slug"] = data.get("model").replace(" ", "-").lower()
                         except AttributeError:
                             raise ProxboxException(
-                                message=f"<span class='badge text-bg-red' title='Post'><strong><i class='mdi mdi-upload'></i></strong></span> No <strong>NAME</strong> or <strong>model</strong> field provided on the payload. Please provide one of them.",
+                                message="<span class='badge text-bg-red' title='Post'><strong><i class='mdi mdi-upload'></i></strong></span> No <strong>NAME</strong> or <strong>model</strong> field provided on the payload. Please provide one of them.",
                             )
 
-        if self.default or data == None:
+        if self.default or data is None:
             await log(self.websocket, f"<span class='badge text-bg-red' title='Post'><strong><i class='mdi mdi-upload'></i></strong></span> Creating DEFAULT <strong>{self.object_name}</strong> object on Netbox.")
-            data = self.base_dict
+            data: dict = self.base_dict
 
         try:
 
             """
             Merge base_dict and data dict.
             The fields not specificied on data dict will be filled with the base_dict values.
             """
-            data = self.base_dict | data
+            data: dict = self.base_dict | data
 
             check_duplicate_result = await self._check_duplicate(object = data)
 
-            if check_duplicate_result == None:
+            if check_duplicate_result is None:
 
                 # Check if tags field exists on the payload and if true, append the Proxbox tag. If not, create it.
-                if data.get("tags") == None:
+                if data.get("tags") is None:
                     data["tags"] = [self.nb.tag.id]
                 else:
                     data["tags"].append(self.nb.tag.id)
@@ -498,7 +509,8 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
                 # create = self.pynetbox_path.create(self.default_dict)
                 # return create
 
-            except ProxboxException as error: raise error
+            except ProxboxException as error:
+                raise error
 
             except Exception as error:
                 raise ProxboxException(
@@ -624,7 +636,7 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
                                 await log(self.websocket, "<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
 
-                        await log(self.websocket, f"[CHECK_DUPLICATE] Object found on Netbox. Returning it.")
+                        await log(self.websocket, "[CHECK_DUPLICATE] Object found on Netbox. Returning it.")
                         print(f'result_by_primary: {result_by_primary}')
                         return result_by_primary
 
@@ -634,7 +646,11 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
                 result = await asyncio.to_thread(self.pynetbox_path.get, dict(object))
 
                 if result:
-                    await log(self.websocket, f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> <strong>Object found</strong> on Netbox. Returning it.")
+                    await log(
+                        self.websocket,
+                        "<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> <strong>Object found</strong> on Netbox. Returning it."
+                    )
+
                     return result
 
                 else:
@@ -659,18 +675,20 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
                         print(f"device_obj.id: {device_obj.id}")
 
                         result_by_device = await asyncio.to_thread(self.pynetbox_path.get,
-                            name=object.get("name"),
+                            name=object.get('name'),
                             tag=[self.nb.tag.slug]
                         )
 
 
-
-                    except:
-                        await log(self.websocket, "<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> (1.5.1) Device Object <strong>NOT</strong> found when checking for duplicated using <strong>Device<strong> as parameter.")
+                    except Exception as error:
+                        await log(
+                            self.websocket,
+                            f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> (1.5.1) Device Object <strong>NOT</strong> found when checking for duplicated using <strong>Device<strong> as parameter.<br>{error}"
+                        )
 
 
                     if result_by_device:
-                        if int(object.get('device')) != int(result_by_device.device.id):
+                        if int(object.get('device', 0)) != int(result_by_device.device.id):
                             return None
 
                         await log(self.websocket, "<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> (1.5.1) <strong>Object found</strong> on Netbox. Returning it.")
@@ -691,6 +709,7 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
                         print(result_by_tag)
 
                     except Exception as error:
+                        print(f'Error: {error}')
 
                         try:
                             result_by_tag = await asyncio.to_thread(self.pynetbox_path.filter,
@@ -710,17 +729,23 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
                                         return obj
                                 return None
                             print(f"filter: {result_by_tag}")
+
                         except Exception as error:
                             logger.error(error)
-
-
 
                     if result_by_tag:
-                        await log(self.websocket, f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> <strong>Object found</strong> on Netbox. Returning it.")
+                        await log(
+                            self.websocket,
+                            "<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> <strong>Object found</strong> on Netbox. Returning it."
+                        )
+
                         return result_by_tag
 
+                    await log(
+                        self.websocket,
+                        "<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> (3) Checking <strong>duplicate object</strong> using only <strong>NAME</strong> and <strong>SLUG</strong>"
+                    )
 
-                    await log(self.websocket, f"<span class='badge text-bg-purple' title='Check Duplicate'><i class='mdi mdi-content-duplicate'></i></span> (3) Checking <strong>duplicate object</strong> using only <strong>NAME</strong> and <strong>SLUG</strong>")
                     result_by_name_and_slug = await asyncio.to_thread(self.pynetbox_path.get,
                         name=object.get("name"),
                         slug=object.get("slug"),
@@ -734,6 +759,7 @@ async def _check_duplicate(self, search_params: dict = None, object: dict = None
 
                 return None
 
-            except ProxboxException as error: raise error
+            except ProxboxException as error:
+                raise error
 
         return None