SG-38306 Python2 Removal - Part 2 - Easy ones

docs/reference.rst

@@ -1034,6 +1034,3 @@ Example for a user whose language preference is set to Japanese:
     },
     ...
     }
-
-.. note::
-    If needed, the encoding of the returned localized string can be ensured regardless the Python version using shotgun_api3.lib.six.ensure_text().

shotgun_api3/shotgun.py

@@ -712,9 +712,9 @@ def __init__(
         # the lowercase version of the credentials.
         auth, self.config.server = self._split_url(base_url)
         if auth:
-            auth = base64encode(
-                sgutils.ensure_binary(urllib.parse.unquote(auth))
-            ).decode("utf-8")
+            auth = base64encode(urllib.parse.unquote(auth).encode("utf-8")).decode(
+                "utf-8"
+            )
             self.config.authorization = "Basic " + auth.strip()
 
         # foo:bar@123.456.789.012:3456
@@ -2270,8 +2270,7 @@ def schema_field_update(
             "type": entity_type,
             "field_name": field_name,
             "properties": [
-                {"property_name": k, "value": v}
-                for k, v in six.iteritems((properties or {}))
+                {"property_name": k, "value": v} for k, v in (properties or {}).items()
             ],
         }
         params = self._add_project_param(params, project_entity)
@@ -2966,7 +2965,11 @@ def download_attachment(self, attachment=False, file_path=None, attachment_id=No
                         url.find("s3.amazonaws.com") != -1
                         and e.headers["content-type"] == "application/xml"
                     ):
-                        body = [sgutils.ensure_text(line) for line in e.readlines()]
+                        body = [
+                            line.decode("utf-8") if isinstance(line, bytes) else line
+                            for line in e.readlines()
+                        ]
+
                         if body:
                             xml = "".join(body)
                             # Once python 2.4 support is not needed we can think about using
@@ -3328,7 +3331,7 @@ def text_search(self, text, entity_types, project_ids=None, limit=None):
             raise ValueError("entity_types parameter must be a dictionary")
 
         api_entity_types = {}
-        for entity_type, filter_list in six.iteritems(entity_types):
+        for entity_type, filter_list in entity_types.items():
 
             if isinstance(filter_list, (list, tuple)):
                 resolved_filters = _translate_filters(filter_list, filter_operator=None)
@@ -3859,8 +3862,7 @@ def _encode_payload(self, payload):
         be in a single byte encoding to go over the wire.
         """
 
-        wire = json.dumps(payload, ensure_ascii=False)
-        return sgutils.ensure_binary(wire)
+        return json.dumps(payload, ensure_ascii=False).encode("utf-8")
 
     def _make_call(self, verb, path, body, headers):
         """
@@ -3965,7 +3967,7 @@ def _http_request(self, verb, path, body, headers):
         resp, content = conn.request(url, method=verb, body=body, headers=headers)
         # http response code is handled else where
         http_status = (resp.status, resp.reason)
-        resp_headers = dict((k.lower(), v) for k, v in six.iteritems(resp))
+        resp_headers = dict((k.lower(), v) for k, v in resp.items())
         resp_body = content
 
         LOG.debug("Response status is %s %s" % http_status)
@@ -4045,7 +4047,7 @@ def _decode_list(lst):
 
         def _decode_dict(dct):
             newdict = {}
-            for k, v in six.iteritems(dct):
+            for k, v in dct.items():
                 if isinstance(k, str):
                     k = sgutils.ensure_str(k)
                 if isinstance(v, str):
@@ -4119,7 +4121,7 @@ def _visit_data(self, data, visitor):
             return tuple(recursive(i, visitor) for i in data)
 
         if isinstance(data, dict):
-            return dict((k, recursive(v, visitor)) for k, v in six.iteritems(data))
+            return dict((k, recursive(v, visitor)) for k, v in data.items())
 
         return visitor(data)
 
@@ -4166,10 +4168,6 @@ def _outbound_visitor(value):
                     value = _change_tz(value)
                 return value.strftime("%Y-%m-%dT%H:%M:%SZ")
 
-            # ensure return is six.text_type
-            if isinstance(value, str):
-                return sgutils.ensure_text(value)
-
             return value
 
         return self._visit_data(data, _outbound_visitor)
@@ -4288,7 +4286,7 @@ def _parse_records(self, records):
                 continue
 
             # iterate over each item and check each field for possible injection
-            for k, v in six.iteritems(rec):
+            for k, v in rec.items():
                 if not v:
                     continue
 
@@ -4376,7 +4374,7 @@ def _dict_to_list(
         [{'field_name': 'foo', 'value': 'bar', 'thing1': 'value1'}]
         """
         ret = []
-        for k, v in six.iteritems((d or {})):
+        for k, v in (d or {}).items():
             d = {key_name: k, value_name: v}
             d.update((extra_data or {}).get(k, {}))
             ret.append(d)
@@ -4389,7 +4387,7 @@ def _dict_to_extra_data(self, d, key_name="value"):
 
         e.g. d {'foo' : 'bar'} changed to {'foo': {"value": 'bar'}]
         """
-        return dict([(k, {key_name: v}) for (k, v) in six.iteritems((d or {}))])
+        return dict([(k, {key_name: v}) for (k, v) in (d or {}).items()])
 
     def _upload_file_to_storage(self, path, storage_url):
         """
@@ -4657,7 +4655,10 @@ def _send_form(self, url, params):
                 else:
                     raise ShotgunError("Unanticipated error occurred %s" % (e))
 
-            return sgutils.ensure_text(result)
+            if isinstance(result, bytes):
+                result = result.decode("utf-8")
+
+            return result
         else:
             raise ShotgunError("Max attemps limit reached.")
 
@@ -4738,9 +4739,8 @@ def http_request(self, request):
                 else:
                     params.append((key, value))
             if not files:
-                data = sgutils.ensure_binary(
-                    urllib.parse.urlencode(params, True)
-                )  # sequencing on
+                data = urllib.parse.urlencode(params, True).encode("utf-8")
+                # sequencing on
             else:
                 boundary, data = self.encode(params, files)
                 content_type = "multipart/form-data; boundary=%s" % boundary
@@ -4763,42 +4763,48 @@ def encode(self, params, files, boundary=None, buffer=None):
         if buffer is None:
             buffer = BytesIO()
         for key, value in params:
-            if not isinstance(value, str):
+            if isinstance(key, bytes):
+                key = key.decode("utf-8")
+
+            if isinstance(value, bytes):
+                value = value.decode("utf-8")
+            elif not isinstance(value, str):
                 # If value is not a string (e.g. int) cast to text
                 value = str(value)
-            value = sgutils.ensure_text(value)
-            key = sgutils.ensure_text(key)
 
-            buffer.write(sgutils.ensure_binary("--%s\r\n" % boundary))
+            buffer.write(f"--{boundary}\r\n".encode("utf-8"))
             buffer.write(
-                sgutils.ensure_binary('Content-Disposition: form-data; name="%s"' % key)
+                f'Content-Disposition: form-data; name="{key}"'.encode("utf-8")
             )
-            buffer.write(sgutils.ensure_binary("\r\n\r\n%s\r\n" % value))
+            buffer.write(f"\r\n\r\n{value}\r\n".encode("utf-8"))
         for key, fd in files:
             # On Windows, it's possible that we were forced to open a file
             # with non-ascii characters as unicode. In that case, we need to
             # encode it as a utf-8 string to remove unicode from the equation.
             # If we don't, the mix of unicode and strings going into the
             # buffer can cause UnicodeEncodeErrors to be raised.
-            filename = fd.name
-            filename = sgutils.ensure_text(filename)
+            filename = (
+                fd.name.decode("utf-8") if isinstance(fd.name, bytes) else fd.name
+            )
             filename = filename.split("/")[-1]
-            key = sgutils.ensure_text(key)
+            if isinstance(key, bytes):
+                key = key.decode("utf-8")
+
             content_type = mimetypes.guess_type(filename)[0]
             content_type = content_type or "application/octet-stream"
             file_size = os.fstat(fd.fileno())[stat.ST_SIZE]
-            buffer.write(sgutils.ensure_binary("--%s\r\n" % boundary))
+            buffer.write(f"--{boundary}\r\n".encode("utf-8"))
             c_dis = 'Content-Disposition: form-data; name="%s"; filename="%s"%s'
             content_disposition = c_dis % (key, filename, "\r\n")
-            buffer.write(sgutils.ensure_binary(content_disposition))
-            buffer.write(sgutils.ensure_binary("Content-Type: %s\r\n" % content_type))
-            buffer.write(sgutils.ensure_binary("Content-Length: %s\r\n" % file_size))
+            buffer.write(content_disposition.encode("utf-8"))
+            buffer.write(f"Content-Type: {content_type}\r\n".encode("utf-8"))
+            buffer.write(f"Content-Length: {file_size}\r\n".encode("utf-8"))
 
-            buffer.write(sgutils.ensure_binary("\r\n"))
+            buffer.write(b"\r\n")
             fd.seek(0)
             shutil.copyfileobj(fd, buffer)
-            buffer.write(sgutils.ensure_binary("\r\n"))
-        buffer.write(sgutils.ensure_binary("--%s--\r\n\r\n" % boundary))
+            buffer.write(b"\r\n")
+        buffer.write(f"--{boundary}--\r\n\r\n".encode("utf-8"))
         buffer = buffer.getvalue()
         return boundary, buffer
 

tests/base.py

@@ -135,7 +135,7 @@ class MockTestBase(TestBase):
     """Test base for tests mocking server interactions."""
 
     def setUp(self):
-        super(MockTestBase, self).setUp()
+        super().setUp()
         # TODO see if there is another way to stop sg connecting
         self._setup_mock()
         self._setup_mock_data()
@@ -252,7 +252,7 @@ class LiveTestBase(TestBase):
     def setUp(self, auth_mode=None):
         if not auth_mode:
             auth_mode = "HumanUser" if self.config.jenkins else "ApiUser"
-        super(LiveTestBase, self).setUp(auth_mode)
+        super().setUp(auth_mode)
         if (
             self.sg.server_caps.version
             and self.sg.server_caps.version >= (3, 3, 0)
@@ -410,7 +410,7 @@ class HumanUserAuthLiveTestBase(LiveTestBase):
     """
 
     def setUp(self):
-        super(HumanUserAuthLiveTestBase, self).setUp("HumanUser")
+        super().setUp("HumanUser")
 
 
 class SessionTokenAuthLiveTestBase(LiveTestBase):
@@ -420,7 +420,7 @@ class SessionTokenAuthLiveTestBase(LiveTestBase):
     """
 
     def setUp(self):
-        super(SessionTokenAuthLiveTestBase, self).setUp("SessionToken")
+        super().setUp("SessionToken")
 
 
 class SgTestConfig(object):