add missing trailing commas

gso/services/librenms.py → gso/services/librenms_client.py

-"""
-The LibreNMS module interacts with the LibreNMS instance when
+"""The LibreNMS module interacts with the LibreNMS instance when
 - Creating a device.
 - Validating the input of a device.
 - Terminating a device.
 """
 import json
 import logging
+
 import requests
-from gso import settings
 
+from gso import settings
 
 logger = logging.getLogger(__name__)
 
 
-class CfgStruct(object):
+class CfgStruct:
     pass
 
 
 def _get_cfg():
-    """
-    Internal function to retrieve all needed configuration.
-    """
+    """Internal function to retrieve all needed configuration."""
     oss = settings.load_oss_params()
     cfg = CfgStruct()
     # Hack for later ease: 1st setattr will fill in the inner's dict
-    setattr(cfg, "_hack", "")
+    cfg._hack = ""
     # Update inner dict
     cfg.__dict__.update(oss.MONITORING)
     assert cfg.__dict__ is not None
@@ -51,8 +49,7 @@ def _get_cfg():
 
 
 def validate_device(fqdn: str):
-    """
-    Function that validates the existence of a device in LibreNMS.
+    """Function that validates the existence of a device in LibreNMS.
 
     :param FQDN of the device to validate.
     """
@@ -63,10 +60,8 @@ def validate_device(fqdn: str):
         CFG.url_devices, headers=CFG.headers)
     assert nms_result is not None
 
-    device_id = list(map(
-        lambda x: x.get("device_id"),
-        filter(lambda x: x.get("hostname") == fqdn,
-               nms_result.json().get("devices"))))
+    device_id = [x.get("device_id") for x in filter(lambda x: x.get("hostname") == fqdn,
+               nms_result.json().get("devices"))]
 
     if len(device_id) != 1 or device_id[0] is None:
         error_msg = f"Device with FQDN={fqdn} is not registered in LibreNMS"
@@ -96,8 +91,7 @@ def validate_device(fqdn: str):
 
 
 def register_device(fqdn: str):
-    """
-    Function that registers a new device in LibreNMS.
+    """Function that registers a new device in LibreNMS.
 
     :param FQDN of the device to register.
     """
@@ -115,7 +109,7 @@ def register_device(fqdn: str):
     }
     if CFG.SNMP.version == "v2c":
         device_data.update({
-            "community": CFG.SNMP.V2.community
+            "community": CFG.SNMP.V2.community,
             })
 
     elif CFG.SNMP.version == "v3":
@@ -138,8 +132,7 @@ def register_device(fqdn: str):
 
 
 def deregister_device(fqdn: str):
-    """
-    Function that reregisters a device from LibreNMS.
+    """Function that reregisters a device from LibreNMS.
 
     :param FQDN of the device to deregister.
     """
@@ -149,10 +142,8 @@ def deregister_device(fqdn: str):
     nms_result = requests.get(
         CFG.url_devices, headers=CFG.headers)
     assert nms_result is not None
-    device_id = list(map(
-        lambda x: x.get("device_id"),
-        filter(lambda x: x.get("hostname") == fqdn,
-               nms_result.json().get("devices"))))
+    device_id = [x.get("device_id") for x in filter(lambda x: x.get("hostname") == fqdn,
+               nms_result.json().get("devices"))]
     if len(device_id) != 1:
         return
     device_id = device_id[0]
@@ -160,7 +151,7 @@ def deregister_device(fqdn: str):
     # https://docs.librenms.org/API/Devices/#endpoint-categories
     device_data = {
         "field": "disabled",
-        "data": "1"
+        "data": "1",
     }
     url_device = f"{CFG.url_devices}/{device_id}"
     logger.debug(f"Connecting to URL: {url_device}"

gso/services/subscriptions.py

@@ -115,7 +115,7 @@ def get_active_trunks_that_terminate_on_router(subscription_id: UUIDstr) -> list
     """
     return query_in_use_by_subscriptions(UUID(subscription_id)).join(ProductTable).filter(
         ProductTable.product_type == "Iptrunk",
-        SubscriptionTable.status == "active"
+        SubscriptionTable.status == "active",
     ).all()
 
 

gso/settings.py

@@ -21,8 +21,6 @@ class GeneralParams(BaseSettings):
     public_hostname: str
     """The hostname that :term:`GSO` is publicly served at, used for building the callback URL that the provisioning
     proxy uses."""
-    environment: str
-    """The environment in which :term:`GSO` runs."""
 
 
 class CeleryParams(BaseSettings):
@@ -97,24 +95,21 @@ class IPAMParams(BaseSettings):
 
 
 class MonitoringLibreNMSDevGroupsParams(BaseSettings):
-    """
-    Parameters related to LibreNMS device groups.
-    """
+    """Parameters related to LibreNMS device groups."""
+
     routers_lab: str
     routers_prod: str
 
 
 class MonitoringSNMPV2Params(BaseSettings):
-    """
-    Parameters related to SNMPv2.
-    """
+    """Parameters related to SNMPv2."""
+
     community: str
 
 
 class MonitoringSNMPV3Params(BaseSettings):
-    """
-    Parameters related to SNMPv3.
-    """
+    """Parameters related to SNMPv3."""
+
     authlevel: str
     authname: str
     authpass: str
@@ -123,30 +118,19 @@ class MonitoringSNMPV3Params(BaseSettings):
     cryptoalgo: str
 
 
-class MonitoringSNMPParams(BaseSettings):
-    """
-    Parameters related to SNMP.
-    """
-    version: str
-    V2: MonitoringSNMPV2Params
-    V3: MonitoringSNMPV3Params
-
-
 class MonitoringLibreNMSParams(BaseSettings):
-    """
-    Parameters related to LibreNMS.
-    """
+    """Parameters related to LibreNMS."""
+
     endpoint: str
     token: str
     DEVICE_GROUPS: MonitoringLibreNMSDevGroupsParams
 
 
 class MonitoringParams(BaseSettings):
-    """
-    Parameters related to the monitoring.
-    """
+    """Parameters related to the monitoring."""
+
     LIBRENMS: MonitoringLibreNMSParams
-    SNMP: MonitoringSNMPParams
+    SNMP: MonitoringSNMPV2Params | MonitoringSNMPV3Params
 
 
 class ProvisioningProxyParams(BaseSettings):

gso/workflows/router/update_ibgp_mesh.py

@@ -10,8 +10,8 @@ from pydantic import root_validator
 
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
-from gso.services import provisioning_proxy, subscriptions
-from gso.services.provisioning_proxy import pp_interaction, indifferent_pp_interaction
+from gso.services import librenms_client, provisioning_proxy, subscriptions
+from gso.services.provisioning_proxy import indifferent_pp_interaction, pp_interaction
 from gso.services.subscriptions import get_active_trunks_that_terminate_on_router
 
 
@@ -52,14 +52,14 @@ def _generate_pe_inventory(pe_router_list: list[Router]) -> dict[str, Any]:
                     "lo4": router.router.router_lo_ipv4_address,
                     "lo6": router.router.router_lo_ipv6_address,
                     "vendor": router.router.vendor,
-                }
+                },
             } for router in pe_router_list
         },
         "all": {
             "hosts": {
                 router.router.router_fqdn: None for router in pe_router_list
-            }
-        }
+            },
+        },
     }
 
 
@@ -67,7 +67,7 @@ def _generate_pe_inventory(pe_router_list: list[Router]) -> dict[str, Any]:
 def add_p_to_mesh_dry(subscription: Router, callback_route: str, pe_router_list: list[Router]) -> State:
     extra_vars = {
         "dry_run": True,
-        "subscription": subscription
+        "subscription": subscription,
     }
 
     provisioning_proxy.execute_playbook(
@@ -84,7 +84,7 @@ def add_p_to_mesh_dry(subscription: Router, callback_route: str, pe_router_list:
 def add_p_to_mesh_real(subscription: Router, callback_route: str, pe_router_list: list[Router]) -> State:
     extra_vars = {
         "dry_run": False,
-        "subscription": subscription
+        "subscription": subscription,
     }
 
     provisioning_proxy.execute_playbook(
@@ -113,9 +113,9 @@ def add_all_pe_to_p_dry(subscription: Router, pe_router_list: list[Router], call
     inventory = {
         "all": {
             "hosts": {
-                router.router.router_fqdn: None
+                router.router.router_fqdn: None,
             } for router in pe_router_list
-        }
+        },
     }
 
     provisioning_proxy.execute_playbook(
@@ -144,9 +144,9 @@ def add_all_pe_to_p_real(subscription: Router, pe_router_list: list[Router], cal
     inventory = {
         "all": {
             "hosts": {
-                router.router.router_fqdn: None
+                router.router.router_fqdn: None,
             } for router in pe_router_list
-        }
+        },
     }
 
     provisioning_proxy.execute_playbook(
@@ -164,9 +164,9 @@ def check_ibgp_session(subscription: Router, callback_route: str) -> State:
     inventory = {
         "all": {
             "hosts": {
-                subscription.router.router_fqdn: None
-            }
-        }
+                subscription.router.router_fqdn: None,
+            },
+        },
     }
 
     provisioning_proxy.execute_playbook(
@@ -199,7 +199,7 @@ def update_subscription_model(subscription: Router) -> State:
     target=Target.MODIFY,
 )
 def update_ibgp_mesh() -> StepList:
-    """Update the iBGP mesh with a new P router
+    """Update the iBGP mesh with a new P router.
 
     * Add the new P-router to all other PE-routers in the network, including a dry run.
     * Add all PE-routers to the P-router, including a dry run.