Fix for Python 3.10

Allow newer versions of requests-toolbelt
Fix all uses of collections

Author: coreone

pytos/common/rest_requests.py

@@ -1,5 +1,8 @@
 
-import collections
+try:
+    from collections import Iterable
+except:
+    from collections.abc import Iterable
 import datetime
 import hashlib
 import http.cookiejar
@@ -222,7 +225,7 @@ def _ensure_response_status(self):
             status_code_ok = False
 
         if status_code_ok:
-            if isinstance(self.expected_status_codes, collections.Iterable):
+            if isinstance(self.expected_status_codes, Iterable):
                 if self.response.status_code not in self.expected_status_codes:
                     status_code_ok = False
             elif isinstance(self.expected_status_codes, int):

pytos/securechange/xml_objects/rest.py

@@ -1,7 +1,11 @@
 import datetime
 import time
 import enum
-from collections import OrderedDict
+try:
+    from collections import OrderedDict
+except:
+    from collections.abc import OrderedDict
+
 
 # For backward compatibility import FileLock
 from pytos.securechange.xml_objects.restapi.step.rule_decommission.rule_decommission import Step_Field_Rule_Decommission
@@ -2132,4 +2136,3 @@ def from_xml_node(cls, xml_node):
 class RejectComment(Comment):
     def __init__(self, comment):
         super().__init__(comment, Elements.REJECT_COMMENT)
-

pytos/securetrack/helpers.py

@@ -1,5 +1,8 @@
 # coding=utf-8
-import collections
+try:
+    from collections import Iterable
+except:
+    from collections.abc import Iterable
 import csv
 import io
 import itertools
@@ -1320,7 +1323,7 @@ def get_rule_documentation_by_device_id_and_rule_id(self, device_id, rule_id):
     def get_rule_base(self, get_documentation=False, device_ids=None):
         """Get the rules for each of the devices configured in SecureTrack.
 
-        :type device_ids: collections.Iterable[int]
+        :type device_ids: Iterable[int]
         :param device_ids: If specified, get the rule base only for the specified devices.
         :param get_documentation: Whether or not to get the rule documentation together with the rule base.
         :type get_documentation: bool
@@ -2140,13 +2143,13 @@ def get_services_by_device_and_object_ids(self, device_id, service_ids):
         :param device_id: The device ID for which we want to get network objects.
         :type device_id: int
         :param service_ids: The ID of the service
-        :type service_ids: int|collections.Iterable[int]
+        :type service_ids: int|Iterable[int]
         :return: The service for the specified device with the specified ID.
         :rtype: Single_Service|Group_Service
         :raise ValueError: If a device with the specified ID does not exist.
         :raise IOError: If there was a communication problem trying to get the network objects.
         """
-        if isinstance(service_ids, collections.Iterable):
+        if isinstance(service_ids, Iterable):
             service_ids = ",".join([str(service_id) for service_id in service_ids])
         logger.info("Getting service with ID %s for device %s.", service_ids, device_id)
         try:
@@ -2169,13 +2172,13 @@ def get_services_by_revision_and_object_ids(self, revision_id, service_ids=""):
         :param revision_id: The revision ID for which we want to get network objects.
         :type revision_id: int
         :param service_ids: The ID of the service
-        :type service_ids: int|collections.Iterable[int]
+        :type service_ids: int|Iterable[int]
         :return: The service for the specified revision with the specified ID.
         :rtype: Services_List
         :raise ValueError: If a revision with the specified ID does not exist.
         :raise IOError: If there was a communication problem trying to get the services.
         """
-        if isinstance(service_ids, collections.Iterable):
+        if isinstance(service_ids, Iterable):
             service_ids = ",".join([str(service_id) for service_id in service_ids])
         logger.info("Getting service with ID %s for revision %s.", service_ids, revision_id)
         try:
@@ -2198,14 +2201,14 @@ def get_network_objects_by_revision_and_object_ids(self, revision_id, network_ob
         :param revision_id: The revision ID for which we want to get network objects.
         :type revision_id: int
         :param network_object_ids: The ID of the network object to get
-        :type network_object_ids: int|collections.Iterable[int]
+        :type network_object_ids: int|Iterable[int]
         :return: The network objects for the specified revision.
         :rtype: Network_Objects_List
         :raise ValueError: If a revision with the specified ID does not exist.
         :raise IOError: Ifp there was a communication problem trying to get the network objects.
         """
         logger.info("Getting network object with ID %s for revision %s.", network_object_ids, revision_id)
-        if isinstance(network_object_ids, collections.Iterable):
+        if isinstance(network_object_ids, Iterable):
             network_object_ids = ",".join([str(network_object_id) for network_object_id in network_object_ids])
         try:
             response_string = self.get_uri(
@@ -2227,14 +2230,14 @@ def get_network_objects_by_device_and_object_ids(self, device_id, network_object
         :param device_id: The device ID for which we want to get network objects.
         :type device_id: int
         :param network_object_ids: The ID of the network object to get
-        :type network_object_ids: int|collections.Iterable[int]
+        :type network_object_ids: int|Iterable[int]
         :return: The network objects for the specified device.
         :rtype: Network_Objects_List
         :raise ValueError: If a device with the specified ID does not exist.
         :raise IOError: Ifp there was a communication problem trying to get the network objects.
         """
         logger.info("Getting network object with ID %s for device %s.", network_object_ids, device_id)
-        if isinstance(network_object_ids, collections.Iterable):
+        if isinstance(network_object_ids, Iterable):
             network_object_ids = ",".join([str(network_object_id) for network_object_id in network_object_ids])
         try:
             response_string = self.get_uri(

setup.py

@@ -40,7 +40,7 @@ def get_version():
         'netaddr>=0.7.14',
         'paramiko>=1.15.2',
         'requests>=2.6.0',
-        'requests_toolbelt==0.7.1',
+        'requests_toolbelt>=0.7.1',
         'netifaces==0.10.9',
         'dnspython3==1.15.0',
         'Mako',