Drop Python 3.7 Support

nautobot_ssot/filters.py

@@ -29,8 +29,12 @@ class Meta:
         model = SyncLogEntry
         fields = ["sync", "action", "status", "synced_object_type"]
 
-    def search(self, queryset, _name, value):  # pylint: disable=no-self-use
+    def search(self, queryset, _name, value):
         """String search of SyncLogEntry records."""
         if not value.strip():
             return queryset
-        return queryset.filter(Q(diff__icontains=value) | Q(message__icontains=value) | Q(object_repr__icontains=value))
+        return queryset.filter(
+            Q(diff__icontains=value)  # pylint: disable=unsupported-binary-operation
+            | Q(message__icontains=value)
+            | Q(object_repr__icontains=value)
+        )

nautobot_ssot/integrations/aci/diffsync/client.py

@@ -216,7 +216,7 @@ def get_static_path(self, tenant: str, ap: str, epg: str) -> list:
         )
         sp_list = []
         for obj in resp.json()["imdata"]:
-            sp_dict = dict(encap=obj["fvRsPathAtt"]["attributes"]["encap"])
+            sp_dict = {"encap": obj["fvRsPathAtt"]["attributes"]["encap"]}
             tDn = obj["fvRsPathAtt"]["attributes"]["tDn"]
             if "paths" in tDn and "protpaths" not in tDn:
                 # port on a single node
@@ -278,25 +278,32 @@ def get_static_path(self, tenant: str, ap: str, epg: str) -> list:
     def get_epg_details(self, tenant: str, ap: str, epg: str) -> dict:
         """Return EPG configuration details."""
         resp = self._get(f"/api/node/mo/uni/tn-{tenant}/ap-{ap}/epg-{epg}.json?query-target=children")
-        epg_dict = dict(bd=None, subnets=[], provided_contracts=[], consumed_contracts=[], domains=[], static_paths=[])
+        epg_dict = {
+            "bd": None,
+            "subnets": [],
+            "provided_contracts": [],
+            "consumed_contracts": [],
+            "domains": [],
+            "static_paths": [],
+        }
         epg_dict["name"] = epg
         for obj in resp.json()["imdata"]:
             if "fvRsBd" in obj:
                 epg_dict["bd"] = obj["fvRsBd"]["attributes"]["tnFvBDName"]
                 epg_dict["subnets"] = self.get_bd_subnet(tenant, epg_dict["bd"])
             if "fvRsCons" in obj:
                 epg_dict["consumed_contracts"].append(
-                    dict(
-                        name=obj["fvRsCons"]["attributes"]["tnVzBrCPName"],
-                        filters=self.get_contract_filters(tenant, obj["fvRsCons"]["attributes"]["tnVzBrCPName"]),
-                    )
+                    {
+                        "name": obj["fvRsCons"]["attributes"]["tnVzBrCPName"],
+                        "filters": self.get_contract_filters(tenant, obj["fvRsCons"]["attributes"]["tnVzBrCPName"]),
+                    }
                 )
             if "fvRsProv" in obj:
                 epg_dict["provided_contracts"].append(
-                    dict(
-                        name=obj["fvRsProv"]["attributes"]["tnVzBrCPName"],
-                        filters=self.get_contract_filters(tenant, obj["fvRsProv"]["attributes"]["tnVzBrCPName"]),
-                    )
+                    {
+                        "name": obj["fvRsProv"]["attributes"]["tnVzBrCPName"],
+                        "filters": self.get_contract_filters(tenant, obj["fvRsProv"]["attributes"]["tnVzBrCPName"]),
+                    }
                 )
             if "fvRsDomAtt" in obj:
                 resp = self._get(f"/api/node/mo/{obj['fvRsDomAtt']['attributes']['tDn']}.json")

nautobot_ssot/integrations/aristacv/utils/cloudvision.py

@@ -79,6 +79,7 @@ def __init__(
                 response = requests.post(  # nosec
                     f"https://{self.cvp_host}/cvpservice/login/authenticate.do",
                     auth=(self.username, self.password),
+                    timeout=60,
                     verify=self.verify,
                 )
                 session_id = response.json().get("sessionId")
@@ -419,7 +420,7 @@ def unfreeze_frozen_dict(frozen_dict):
         dict|str|list: Unfrozen contents of FrozenDict that was passed in.
     """
     if isinstance(frozen_dict, (dict, FrozenDict)):
-        return dict({k: unfreeze_frozen_dict(v) for k, v in frozen_dict.items()})
+        return {k: unfreeze_frozen_dict(v) for k, v in frozen_dict.items()}
 
     if isinstance(frozen_dict, (str)):
         return frozen_dict

nautobot_ssot/integrations/infoblox/utils/client.py

@@ -138,10 +138,10 @@ def _request(self, method, path, **kwargs):
         url = urljoin(self.url, api_path)
 
         if self.cookie:
-            resp = requests.request(method, url, cookies=self.cookie, **kwargs)
+            resp = requests.request(method, url, cookies=self.cookie, timeout=60, **kwargs)
         else:
             kwargs["auth"] = requests.auth.HTTPBasicAuth(self.username, self.password)
-            resp = requests.request(method, url, **kwargs)
+            resp = requests.request(method, url, timeout=60, **kwargs)
             self.cookie = copy.copy(resp.cookies.get_dict("ibapauth"))
         resp.raise_for_status()
         return resp

nautobot_ssot/integrations/ipfabric/diffsync/diffsync_models.py

@@ -301,7 +301,7 @@ def create(cls, diffsync, ids, attrs):
             attrs["mac_address"] = DEFAULT_INTERFACE_MAC
         interface_obj = tonb_nbutils.create_interface(
             device_obj=device_obj,
-            interface_details=dict(**ids, **attrs),
+            interface_details={**ids, **attrs},
         )
         ip_address = attrs["ip_address"]
         if ip_address:

nautobot_ssot/integrations/servicenow/diffsync/adapter_servicenow.py

@@ -237,7 +237,7 @@ def bulk_create_interfaces(self):
 
             for inner_request_id, interface in enumerate(self.interfaces_to_create_per_device[device_name]):
                 inner_request_payload = interface.map_data_to_sn_record(
-                    data=dict(**interface.get_identifiers(), **interface.get_attrs()),
+                    data={**interface.get_identifiers(), **interface.get_attrs()},
                     mapping_entry=sn_mapping_entry,
                 )
                 inner_request_body = b64encode(json.dumps(inner_request_payload).encode("utf-8")).decode("utf-8")

nautobot_ssot/integrations/servicenow/diffsync/models.py

@@ -56,7 +56,7 @@ def create(cls, diffsync, ids, attrs):
         model = super().create(diffsync, ids=ids, attrs=attrs)
 
         sn_resource = diffsync.client.resource(api_path=f"/table/{entry['table']}")
-        sn_record = model.map_data_to_sn_record(data=dict(**ids, **attrs), mapping_entry=entry)
+        sn_record = model.map_data_to_sn_record(data={**ids, **attrs}, mapping_entry=entry)
         sn_resource.create(payload=sn_record)
 
         return model

nautobot_ssot/integrations/servicenow/models.py

@@ -37,6 +37,6 @@ def __str__(self):
         """String representation of singleton instance."""
         return "SSoT ServiceNow Configuration"
 
-    def get_absolute_url(self):  # pylint: disable=no-self-use
+    def get_absolute_url(self):
         """Get URL for the associated configuration view."""
         return reverse("plugins:nautobot_ssot:servicenow_config")

nautobot_ssot/integrations/servicenow/third_party/pysnow/oauth_client.py

@@ -24,10 +24,7 @@ class OAuthClient(Client):
 
     token = None
 
-    def __init__(
-        self, client_id=None, client_secret=None, token_updater=None, **kwargs
-    ):
-
+    def __init__(self, client_id=None, client_secret=None, token_updater=None, **kwargs):
         if not (client_secret and client_id):
             raise InvalidUsage("You must supply a client_id and client_secret")
 
@@ -88,14 +85,10 @@ def set_token(self, token):
             "expires_at",
         ]
         if not isinstance(token, dict) or not set(token) >= set(expected_keys):
-            raise InvalidUsage(
-                "Expected a token dictionary containing the following keys: {0}".format(
-                    expected_keys
-                )
-            )
+            raise InvalidUsage("Expected a token dictionary containing the following keys: {0}".format(expected_keys))
 
         # Set sanitized token
-        self.token = dict((k, v) for k, v in token.items() if k in expected_keys)
+        self.token = {(k, v) for k, v in token.items() if k in expected_keys}
 
     def _legacy_request(self, *args, **kwargs):
         """Makes sure token has been set, then calls parent to create a new :class:`pysnow.LegacyRequest` object
@@ -112,9 +105,7 @@ def _legacy_request(self, *args, **kwargs):
             self.session = self._get_oauth_session()
             return super(OAuthClient, self)._legacy_request(*args, **kwargs)
 
-        raise MissingToken(
-            "You must set_token() before creating a legacy request with OAuthClient"
-        )
+        raise MissingToken("You must set_token() before creating a legacy request with OAuthClient")
 
     def resource(self, api_path=None, base_path="/api/now", chunk_size=None):
         """Overrides :meth:`resource` provided by :class:`pysnow.Client` with extras for OAuth
@@ -132,9 +123,7 @@ def resource(self, api_path=None, base_path="/api/now", chunk_size=None):
             self.session = self._get_oauth_session()
             return super(OAuthClient, self).resource(api_path, base_path, chunk_size)
 
-        raise MissingToken(
-            "You must set_token() before creating a resource with OAuthClient"
-        )
+        raise MissingToken("You must set_token() before creating a resource with OAuthClient")
 
     def generate_token(self, user, password):
         """Takes user and password credentials and generates a new token
@@ -149,19 +138,15 @@ def generate_token(self, user, password):
 
         logger.debug("(TOKEN_CREATE) :: User: %s" % user)
 
-        session = OAuth2Session(
-            client=LegacyApplicationClient(client_id=self.client_id)
-        )
+        session = OAuth2Session(client=LegacyApplicationClient(client_id=self.client_id))
 
         try:
-            return dict(
-                session.fetch_token(
-                    token_url=self.token_url,
-                    username=user,
-                    password=password,
-                    client_id=self.client_id,
-                    client_secret=self.client_secret,
-                )
+            return session.fetch_token(
+                token_url=self.token_url,
+                username=user,
+                password=password,
+                client_id=self.client_id,
+                client_secret=self.client_secret,
             )
         except OAuth2Error as exception:
             raise TokenCreateError(

nautobot_ssot/integrations/servicenow/views.py

@@ -14,7 +14,7 @@ class SSOTServiceNowConfigView(UpdateView):
     form_class = SSOTServiceNowConfigForm
     template_name = "nautobot_ssot_servicenow/config.html"
 
-    def get_object(self, queryset=None):  # pylint: disable=unused-argument,no-self-use
+    def get_object(self, queryset=None):  # pylint: disable=unused-argument
         """Retrieve the SSOTServiceNowConfig singleton instance."""
         return SSOTServiceNowConfig.load()
 

nautobot_ssot/jobs/__init__.py

@@ -16,11 +16,20 @@
     jobs = [ExampleDataSource, ExampleDataTarget]
 
 
+class JobException(Exception):
+    """Exception raised when failure loading integration Job."""
+
+    def __init__(self, message):
+        """Populate exception information."""
+        self.message = message
+        super().__init__(self.message)
+
+
 def _add_integrations():
     for module in each_enabled_integration_module("jobs"):
         for job in module.jobs:
             if job in jobs:
-                raise Exception(f"Job {job} already exists in jobs list for integration {module.__file__}.")
+                raise JobException(message=f"Job {job} already exists in jobs list for integration {module.__file__}.")
             logger.debug("Registering job %s from %s", job, module.__file__)
             jobs.append(job)
 

nautobot_ssot/jobs/base.py

@@ -180,7 +180,7 @@ def record_memory_trace(step: str):
             if self.kwargs["memory_profiling"]:
                 record_memory_trace("sync")
 
-    # pylint: disable-next=no-self-use,unused-argument
+    # pylint: disable=unused-argument
     def lookup_object(self, model_name: str, unique_id: str) -> Union[models.Model, None]:
         """Look up the Nautobot record, if any, identified by the args.
 

nautobot_ssot/jobs/examples.py

@@ -308,12 +308,12 @@ def __init__(self, *args, url=None, token=None, job=None, **kwargs):
 
     def _get_api_data(self, url_path: str) -> Mapping:
         """Returns data from a url_path using pagination."""
-        response = requests.get(f"{self.url}/{url_path}", headers=self.headers, params={"limit": 0})
+        response = requests.get(f"{self.url}/{url_path}", headers=self.headers, params={"limit": 0}, timeout=60)
         response.raise_for_status()
         data = response.json()
         result_data = data["results"]
         while data["next"]:
-            data = requests.get(data["next"], headers=self.headers, params={"limit": 0}).json()
+            data = requests.get(data["next"], headers=self.headers, params={"limit": 0}, timeout=60).json()
             result_data.extend(data["results"])
         return result_data
 
@@ -363,19 +363,19 @@ def load(self):
 
     def post(self, path, data):
         """Send an appropriately constructed HTTP POST request."""
-        response = requests.post(f"{self.url}{path}", headers=self.headers, json=data)
+        response = requests.post(f"{self.url}{path}", headers=self.headers, timeout=60, json=data)
         response.raise_for_status()
         return response
 
     def patch(self, path, data):
         """Send an appropriately constructed HTTP PATCH request."""
-        response = requests.patch(f"{self.url}{path}", headers=self.headers, json=data)
+        response = requests.patch(f"{self.url}{path}", headers=self.headers, timeout=60, json=data)
         response.raise_for_status()
         return response
 
     def delete(self, path):
         """Send an appropriately constructed HTTP DELETE request."""
-        response = requests.delete(f"{self.url}{path}", headers=self.headers)
+        response = requests.delete(f"{self.url}{path}", headers=self.headers, timeout=60)
         response.raise_for_status()
         return response
 
@@ -408,7 +408,9 @@ class ExampleDataSource(DataSource, Job):
     def __init__(self):
         """Initialize ExampleDataSource."""
         super().__init__()
-        self.diffsync_flags = self.diffsync_flags | DiffSyncFlags.SKIP_UNMATCHED_DST
+        self.diffsync_flags = (
+            self.diffsync_flags | DiffSyncFlags.SKIP_UNMATCHED_DST  # pylint: disable=unsupported-binary-operation
+        )
 
     class Meta:
         """Metaclass attributes of ExampleDataSource."""
@@ -469,7 +471,9 @@ class ExampleDataTarget(DataTarget, Job):
     def __init__(self):
         """Initialize ExampleDataTarget."""
         super().__init__()
-        self.diffsync_flags = self.diffsync_flags | DiffSyncFlags.SKIP_UNMATCHED_DST
+        self.diffsync_flags = (
+            self.diffsync_flags | DiffSyncFlags.SKIP_UNMATCHED_DST  # pylint: disable=unsupported-binary-operation
+        )
 
     class Meta:
         """Metaclass attributes of ExampleDataTarget."""

nautobot_ssot/tests/aci/test_api.py

@@ -1,5 +1,5 @@
 """Tests for API"""
-# pylint: disable=no-self-use, import-outside-toplevel, invalid-name
+# pylint: disable=import-outside-toplevel, invalid-name
 import unittest
 from unittest.mock import patch, Mock
 from nautobot_ssot.integrations.aci.diffsync.client import AciApi, RequestHTTPError

nautobot_ssot/tests/aristacv/test_utils_cloudvision.py

@@ -143,7 +143,7 @@ def test_unfreeze_frozen_dict(self):
         frozen_result = cloudvision.unfreeze_frozen_dict(frozen_dict=(test_dict, test_frozen))
         self.assertEqual(frozen_result, [{"test": "test"}, {"test2": "test2"}])
 
-        set_result = cloudvision.unfreeze_frozen_dict(frozen_dict=("test"))
+        set_result = cloudvision.unfreeze_frozen_dict(frozen_dict="test")
         self.assertEqual(set_result, ("test"))
 
     def test_get_device_type_modular(self):

nautobot_ssot/tests/infoblox/test_client.py

@@ -112,7 +112,7 @@ def check_headers(request):
             return True
 
         with requests_mock.Mocker() as req:
-            req.get(f"{LOCALHOST}/test_url", additional_matcher=(check_headers))
+            req.get(f"{LOCALHOST}/test_url", additional_matcher=check_headers)
             resp = self.infoblox_client._request("GET", "test_url")
         self.assertEqual(resp.status_code, 200)
 

nautobot_ssot/tests/servicenow/test_adapter_servicenow.py

@@ -14,11 +14,11 @@
 class MockServiceNowClient:
     """Mock version of the ServiceNowClient class using canned data."""
 
-    def get_by_sys_id(self, table, sys_id):  # pylint: disable=unused-argument,no-self-use
+    def get_by_sys_id(self, table, sys_id):  # pylint: disable=unused-argument
         """Get a record with a given sys_id from a given table."""
         return None
 
-    def all_table_entries(self, table, query=None):  # pylint: disable=no-self-use
+    def all_table_entries(self, table, query=None):
         """Iterator over all records in a given table."""
 
         if table == "cmn_location":