SG-29997 - FIRST PHASE Python2 removing (#352)

* ensure_* functions ans six types replaced by native and sgutils

* improvement for ensure_str function

Author: eduardoChaucaGallegos

shotgun_api3/lib/mockgun/mockgun.py

@@ -120,7 +120,6 @@
 from ...shotgun import _Config
 from .errors import MockgunError
 from .schema import SchemaFactory
-from .. import six
 
 # ----------------------------------------------------------------------------
 # Version
@@ -505,14 +504,14 @@ def _validate_entity_data(self, entity_type, data):
                                    "float": float,
                                    "checkbox": bool,
                                    "percent": int,
-                                   "text": six.string_types,
+                                   "text": str,
                                    "serializable": dict,
-                                   "entity_type": six.string_types,
-                                   "date": six.string_types,
+                                   "entity_type": str,
+                                   "date": str,
                                    "date_time": datetime.datetime,
                                    "duration": int,
-                                   "list": six.string_types,
-                                   "status_list": six.string_types,
+                                   "list": str,
+                                   "status_list": str,
                                    "url": dict}[sg_type]
                 except KeyError:
                     raise ShotgunError(

shotgun_api3/lib/sgutils.py

@@ -0,0 +1,62 @@
+"""
+ -----------------------------------------------------------------------------
+ Copyright (c) 2009-2024, Shotgun Software Inc.
+
+ Redistribution and use in source and binary forms, with or without
+ modification, are permitted provided that the following conditions are met:
+
+  - Redistributions of source code must retain the above copyright notice, this
+    list of conditions and the following disclaimer.
+
+  - Redistributions in binary form must reproduce the above copyright notice,
+    this list of conditions and the following disclaimer in the documentation
+    and/or other materials provided with the distribution.
+
+  - Neither the name of the Shotgun Software Inc nor the names of its
+    contributors may be used to endorse or promote products derived from this
+    software without specific prior written permission.
+
+ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+"""
+
+
+def ensure_binary(s, encoding='utf-8', errors='strict'):
+    """
+    Coerce **s** to bytes.
+
+      - `str` -> encoded to `bytes`
+      - `bytes` -> `bytes`
+    """
+    if isinstance(s, str):
+        return s.encode(encoding, errors)
+    elif isinstance(s, bytes):
+        return s
+    else:
+        raise TypeError(f"not expecting type '{type(s)}'")
+    
+
+def ensure_str(s, encoding='utf-8', errors='strict'):
+    """Coerce *s* to `str`.
+
+      - `str` -> `str`
+      - `bytes` -> decoded to `str`
+    """
+    if isinstance(s, str):
+        return s
+
+    elif isinstance(s, bytes): 
+        return s.decode(encoding, errors)
+ 
+    raise TypeError(f"not expecting type '{type(s)}'")
+
+
+ensure_text = ensure_str

shotgun_api3/shotgun.py

@@ -32,6 +32,7 @@
 # Python 2/3 compatibility
 from .lib import six
 from .lib import sgsix
+from .lib import sgutils
 from .lib.six import BytesIO               # used for attachment upload
 from .lib.six.moves import map
 
@@ -665,7 +666,7 @@ def __init__(self,
         # the lowercase version of the credentials.
         auth, self.config.server = self._split_url(base_url)
         if auth:
-            auth = base64encode(six.ensure_binary(
+            auth = base64encode(sgutils.ensure_binary(
                 urllib.parse.unquote(auth))).decode("utf-8")
             self.config.authorization = "Basic " + auth.strip()
 
@@ -2440,7 +2441,7 @@ def upload(self, entity_type, entity_id, path, field_name=None, display_name=Non
         # have to raise a sane exception. This will always work for ascii and utf-8
         # encoded strings, but will fail on some others if the string includes non
         # ascii characters.
-        if not isinstance(path, six.text_type):
+        if not isinstance(path, str):
             try:
                 path = path.decode("utf-8")
             except UnicodeDecodeError:
@@ -2721,7 +2722,7 @@ def download_attachment(self, attachment=False, file_path=None, attachment_id=No
                 elif e.code == 403:
                     # Only parse the body if it is an Amazon S3 url.
                     if url.find("s3.amazonaws.com") != -1 and e.headers["content-type"] == "application/xml":
-                        body = [six.ensure_text(line) for line in e.readlines()]
+                        body = [sgutils.ensure_text(line) for line in e.readlines()]
                         if body:
                             xml = "".join(body)
                             # Once python 2.4 support is not needed we can think about using
@@ -3545,7 +3546,7 @@ def _encode_payload(self, payload):
         """
 
         wire = json.dumps(payload, ensure_ascii=False)
-        return six.ensure_binary(wire)
+        return sgutils.ensure_binary(wire)
 
     def _make_call(self, verb, path, body, headers):
         """
@@ -3720,8 +3721,8 @@ def _json_loads_ascii(self, body):
         def _decode_list(lst):
             newlist = []
             for i in lst:
-                if isinstance(i, six.text_type):
-                    i = six.ensure_str(i)
+                if isinstance(i, str):
+                    i = sgutils.ensure_str(i)
                 elif isinstance(i, list):
                     i = _decode_list(i)
                 newlist.append(i)
@@ -3730,10 +3731,10 @@ def _decode_list(lst):
         def _decode_dict(dct):
             newdict = {}
             for k, v in six.iteritems(dct):
-                if isinstance(k, six.text_type):
-                    k = six.ensure_str(k)
-                if isinstance(v, six.text_type):
-                    v = six.ensure_str(v)
+                if isinstance(k, str):
+                    k = sgutils.ensure_str(k)
+                if isinstance(v, str):
+                    v = sgutils.ensure_str(v)
                 elif isinstance(v, list):
                     v = _decode_list(v)
                 newdict[k] = v
@@ -3844,8 +3845,8 @@ def _outbound_visitor(value):
                 return value.strftime("%Y-%m-%dT%H:%M:%SZ")
 
             # ensure return is six.text_type
-            if isinstance(value, six.string_types):
-                return six.ensure_text(value)
+            if isinstance(value, str):
+                return sgutils.ensure_text(value)
 
             return value
 
@@ -3865,7 +3866,7 @@ def _change_tz(x):
             _change_tz = None
 
         def _inbound_visitor(value):
-            if isinstance(value, six.string_types):
+            if isinstance(value, str):
                 if len(value) == 20 and self._DATE_TIME_PATTERN.match(value):
                     try:
                         # strptime was not on datetime in python2.4
@@ -4266,7 +4267,7 @@ def _send_form(self, url, params):
                 else:
                     raise ShotgunError("Unanticipated error occurred %s" % (e))
 
-            return six.ensure_text(result)
+            return sgutils.ensure_text(result)
         else:
             raise ShotgunError("Max attemps limit reached.")
 
@@ -4339,7 +4340,7 @@ def http_request(self, request):
             data = request.data
         else:
             data = request.get_data()
-        if data is not None and not isinstance(data, six.string_types):
+        if data is not None and not isinstance(data, str):
             files = []
             params = []
             for key, value in data.items():
@@ -4348,7 +4349,7 @@ def http_request(self, request):
                 else:
                     params.append((key, value))
             if not files:
-                data = six.ensure_binary(urllib.parse.urlencode(params, True))  # sequencing on
+                data = sgutils.ensure_binary(urllib.parse.urlencode(params, True))  # sequencing on
             else:
                 boundary, data = self.encode(params, files)
                 content_type = "multipart/form-data; boundary=%s" % boundary
@@ -4371,40 +4372,40 @@ def encode(self, params, files, boundary=None, buffer=None):
         if buffer is None:
             buffer = BytesIO()
         for (key, value) in params:
-            if not isinstance(value, six.string_types):
+            if not isinstance(value, str):
                 # If value is not a string (e.g. int) cast to text
-                value = six.text_type(value)
-            value = six.ensure_text(value)
-            key = six.ensure_text(key)
+                value = str(value)
+            value = sgutils.ensure_text(value)
+            key = sgutils.ensure_text(key)
 
-            buffer.write(six.ensure_binary("--%s\r\n" % boundary))
-            buffer.write(six.ensure_binary("Content-Disposition: form-data; name=\"%s\"" % key))
-            buffer.write(six.ensure_binary("\r\n\r\n%s\r\n" % value))
+            buffer.write(sgutils.ensure_binary("--%s\r\n" % boundary))
+            buffer.write(sgutils.ensure_binary("Content-Disposition: form-data; name=\"%s\"" % key))
+            buffer.write(sgutils.ensure_binary("\r\n\r\n%s\r\n" % value))
         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 = six.ensure_text(filename)
+            filename = sgutils.ensure_text(filename)
             filename = filename.split("/")[-1]
-            key = six.ensure_text(key)
+            key = sgutils.ensure_text(key)
             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(six.ensure_binary("--%s\r\n" % boundary))
+            buffer.write(sgutils.ensure_binary("--%s\r\n" % boundary))
             c_dis = "Content-Disposition: form-data; name=\"%s\"; filename=\"%s\"%s"
             content_disposition = c_dis % (key, filename, "\r\n")
-            buffer.write(six.ensure_binary(content_disposition))
-            buffer.write(six.ensure_binary("Content-Type: %s\r\n" % content_type))
-            buffer.write(six.ensure_binary("Content-Length: %s\r\n" % file_size))
+            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(six.ensure_binary("\r\n"))
+            buffer.write(sgutils.ensure_binary("\r\n"))
             fd.seek(0)
             shutil.copyfileobj(fd, buffer)
-            buffer.write(six.ensure_binary("\r\n"))
-        buffer.write(six.ensure_binary("--%s--\r\n\r\n" % boundary))
+            buffer.write(sgutils.ensure_binary("\r\n"))
+        buffer.write(sgutils.ensure_binary("--%s--\r\n\r\n" % boundary))
         buffer = buffer.getvalue()
         return boundary, buffer
 

tests/base.py

@@ -175,7 +175,7 @@ def _mock_http(self, data, headers=None, status=None):
         if not isinstance(self.sg._http_request, mock.Mock):
             return
 
-        if not isinstance(data, six.string_types):
+        if not isinstance(data, str):
             if six.PY2:
                 data = json.dumps(
                     data,
@@ -208,7 +208,7 @@ def _assert_http_method(self, method, params, check_auth=True):
         """Asserts _http_request is called with the method and params."""
         args, _ = self.sg._http_request.call_args
         arg_body = args[2]
-        assert isinstance(arg_body, six.binary_type)
+        assert isinstance(arg_body, bytes)
         arg_body = json.loads(arg_body)
 
         arg_params = arg_body.get("params")

tests/test_client.py

@@ -17,7 +17,7 @@
 import re
 
 from shotgun_api3.lib.six.moves import urllib
-from shotgun_api3.lib import six
+from shotgun_api3.lib import six, sgutils
 try:
     import simplejson as json
 except ImportError:
@@ -44,7 +44,7 @@
 
 
 def b64encode(val):
-    return base64encode(six.ensure_binary(val)).decode("utf-8")
+    return base64encode(sgutils.ensure_binary(val)).decode("utf-8")
 
 
 class TestShotgunClient(base.MockTestBase):
@@ -424,7 +424,7 @@ def test_call_rpc(self):
 
         # Test unicode mixed with utf-8 as reported in Ticket #17959
         d = {"results": ["foo", "bar"]}
-        a = {"utf_str": "\xe2\x88\x9a", "unicode_str": six.ensure_text("\xe2\x88\x9a")}
+        a = {"utf_str": "\xe2\x88\x9a", "unicode_str": sgutils.ensure_text("\xe2\x88\x9a")}
         self._mock_http(d)
         rv = self.sg._call_rpc("list", a)
         expected = "rpc response with list result"
@@ -585,14 +585,14 @@ def _datetime(s, f):
             return datetime.datetime(*time.strptime(s, f)[:6])
 
         def assert_wire(wire, match):
-            self.assertTrue(isinstance(wire["date"], six.string_types))
+            self.assertTrue(isinstance(wire["date"], str))
             d = _datetime(wire["date"], "%Y-%m-%d").date()
             d = wire['date']
             self.assertEqual(match["date"], d)
-            self.assertTrue(isinstance(wire["datetime"], six.string_types))
+            self.assertTrue(isinstance(wire["datetime"], str))
             d = _datetime(wire["datetime"], "%Y-%m-%dT%H:%M:%SZ")
             self.assertEqual(match["datetime"], d)
-            self.assertTrue(isinstance(wire["time"], six.string_types))
+            self.assertTrue(isinstance(wire["time"], str))
             d = _datetime(wire["time"], "%Y-%m-%dT%H:%M:%SZ")
             self.assertEqual(match["time"], d.time())
 
@@ -621,16 +621,16 @@ def test_encode_payload(self):
 
         d = {"this is ": u"my data \u00E0"}
         j = self.sg._encode_payload(d)
-        self.assertTrue(isinstance(j, six.binary_type))
+        self.assertTrue(isinstance(j, bytes))
 
         d = {
             "this is ": u"my data"
         }
         j = self.sg._encode_payload(d)
-        self.assertTrue(isinstance(j, six.binary_type))
+        self.assertTrue(isinstance(j, bytes))
 
     def test_decode_response_ascii(self):
-        self._assert_decode_resonse(True, six.ensure_str(u"my data \u00E0", encoding='utf8'))
+        self._assert_decode_resonse(True, sgutils.ensure_str(u"my data \u00E0", encoding='utf8'))
 
     def test_decode_response_unicode(self):
         self._assert_decode_resonse(False, u"my data \u00E0")