diff --git a/shotgun_api3/lib/mockgun/mockgun.py b/shotgun_api3/lib/mockgun/mockgun.py index 4bf1e4bf..36b98dd5 100644 --- a/shotgun_api3/lib/mockgun/mockgun.py +++ b/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( diff --git a/shotgun_api3/lib/sgutils.py b/shotgun_api3/lib/sgutils.py new file mode 100644 index 00000000..0d49e4b3 --- /dev/null +++ b/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 diff --git a/shotgun_api3/shotgun.py b/shotgun_api3/shotgun.py index 4d65be51..f5c2c520 100644 --- a/shotgun_api3/shotgun.py +++ b/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,15 +4372,15 @@ 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 @@ -4387,24 +4388,24 @@ def encode(self, params, files, boundary=None, buffer=None): # 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 diff --git a/tests/base.py b/tests/base.py index 04e52aa7..4eaafb86 100644 --- a/tests/base.py +++ b/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") diff --git a/tests/test_client.py b/tests/test_client.py index 6275e4d2..dc3fa3ec 100644 --- a/tests/test_client.py +++ b/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")