Use six instead of twisted.python.compat

src/treq/client.py

@@ -3,12 +3,16 @@
 import mimetypes
 import uuid
 
-from io import BytesIO
+import io
+
+import six
+from six.moves.collections_abc import Mapping
+from six.moves.http_cookiejar import CookieJar
+from six.moves.urllib.parse import urlencode as _urlencode
 
 from twisted.internet.interfaces import IProtocol
 from twisted.internet.defer import Deferred
 from twisted.python.components import proxyForInterface
-from twisted.python.compat import _PY3, unicode
 from twisted.python.filepath import FilePath
 from hyperlink import DecodedURL, EncodedURL
 
@@ -33,25 +37,9 @@
 from treq.response import _Response
 from requests.cookies import cookiejar_from_dict, merge_cookies
 
-if _PY3:
-    from urllib.parse import urlencode as _urlencode
-
-    def urlencode(query, doseq):
-        return _urlencode(query, doseq).encode('ascii')
-    from http.cookiejar import CookieJar
-else:
-    from cookielib import CookieJar
-    from urllib import urlencode
 
-try:
-    # The old location was quixotically deprecated and might actually be
-    # removed in 3.10, maybe.
-    #
-    # See https://github.com/html5lib/html5lib-python/issues/419 for more of
-    # this tale of woe.
-    from collections.abc import Mapping
-except ImportError:
-    from collections import Mapping
+def urlencode(query, doseq):
+    return six.ensure_binary(_urlencode(query, doseq), encoding='ascii')
 
 
 class _BodyBufferingProtocol(proxyForInterface(IProtocol)):
@@ -158,7 +146,7 @@ def request(self, method, url, **kwargs):
             parsed_url = url
         elif isinstance(url, EncodedURL):
             parsed_url = DecodedURL(url)
-        elif isinstance(url, unicode):
+        elif isinstance(url, six.text_type):
             parsed_url = DecodedURL.from_text(url)
         else:
             parsed_url = DecodedURL.from_text(url.decode('ascii'))
@@ -180,7 +168,7 @@ def request(self, method, url, **kwargs):
             if isinstance(headers, dict):
                 h = Headers({})
                 for k, v in headers.items():
-                    if isinstance(v, (bytes, unicode)):
+                    if isinstance(v, (bytes, six.text_type)):
                         h.addRawHeader(k, v)
                     elif isinstance(v, list):
                         h.setRawHeaders(k, v)
@@ -346,20 +334,20 @@ def _coerced_query_params(params):
     for key, values in items:
         if isinstance(key, bytes):
             key = key.decode('ascii')
-        elif not isinstance(key, unicode):
-            key = unicode(key)
+        elif not isinstance(key, six.text_type):
+            key = six.text_type(key)
         if not isinstance(values, (list, tuple)):
             values = [values]
         for value in values:
             if isinstance(value, bytes):
                 value = value.decode('ascii')
-            elif not isinstance(value, unicode):
-                value = unicode(value)
+            elif not isinstance(value, six.text_type):
+                value = six.text_type(value)
             yield key, value
 
 
 def _from_bytes(orig_bytes):
-    return FileBodyProducer(BytesIO(orig_bytes))
+    return FileBodyProducer(io.BytesIO(orig_bytes))
 
 
 def _from_file(orig_file):
@@ -375,14 +363,12 @@ def _guess_content_type(filename):
 
 
 registerAdapter(_from_bytes, bytes, IBodyProducer)
-registerAdapter(_from_file, BytesIO, IBodyProducer)
+registerAdapter(_from_file, io.BytesIO, IBodyProducer)
 
-if not _PY3:
-    from StringIO import StringIO
-    registerAdapter(_from_file, StringIO, IBodyProducer)
+if six.PY2:
+    registerAdapter(_from_file, six.StringIO, IBodyProducer)
     # Suppress lint failure on Python 3.
     registerAdapter(_from_file, file, IBodyProducer)  # noqa: F821
 else:
-    import io
     # file()/open() equiv on Py3
     registerAdapter(_from_file, io.BufferedReader, IBodyProducer)

src/treq/multipart.py

@@ -7,15 +7,13 @@
 from io import BytesIO
 from contextlib import closing
 
+from six import integer_types, text_type
+
 from twisted.internet import defer, task
-from twisted.python.compat import unicode, _PY3
 from twisted.web.iweb import UNKNOWN_LENGTH, IBodyProducer
 
 from zope.interface import implementer
 
-if _PY3:
-    long = int
-
 CRLF = b"\r\n"
 
 
@@ -62,7 +60,7 @@ def __init__(self, fields, boundary=None, cooperator=task):
 
         self.boundary = boundary or uuid4().hex
 
-        if isinstance(self.boundary, unicode):
+        if isinstance(self.boundary, text_type):
             self.boundary = self.boundary.encode('ascii')
 
         self.length = self._calculateLength()
@@ -171,7 +169,7 @@ def _writeLoop(self, consumer):
         consumer.write(CRLF + self._getBoundary(final=True) + CRLF)
 
     def _writeField(self, name, value, consumer):
-        if isinstance(value, unicode):
+        if isinstance(value, text_type):
             self._writeString(name, value, consumer)
         elif isinstance(value, tuple):
             filename, content_type, producer = value
@@ -220,8 +218,8 @@ def _escape(value):
     a newline in the file name parameter makes form-data request unreadable
     for majority of parsers.
     """
-    if not isinstance(value, (bytes, unicode)):
-        value = unicode(value)
+    if not isinstance(value, (bytes, text_type)):
+        value = text_type(value)
     if isinstance(value, bytes):
         value = value.decode('utf-8')
     return value.replace(u"\r", u"").replace(u"\n", u"").replace(u'"', u'\\"')
@@ -234,14 +232,14 @@ def _enforce_unicode(value):
     If someone needs to pass the binary string, use BytesIO and wrap it with
     `FileBodyProducer`.
     """
-    if isinstance(value, unicode):
+    if isinstance(value, text_type):
         return value
 
     elif isinstance(value, bytes):
         # we got a byte string, and we have no ide what's the encoding of it
         # we can only assume that it's something cool
         try:
-            return unicode(value, "utf-8")
+            return text_type(value, "utf-8")
         except UnicodeDecodeError:
             raise ValueError(
                 "Supplied raw bytes that are not ascii/utf-8."
@@ -269,7 +267,7 @@ def _converted(fields):
             filename = _enforce_unicode(filename) if filename else None
             yield name, (filename, content_type, producer)
 
-        elif isinstance(value, (bytes, unicode)):
+        elif isinstance(value, (bytes, text_type)):
             yield name, _enforce_unicode(value)
 
         else:
@@ -302,7 +300,7 @@ def write(self, value):
 
         if value is UNKNOWN_LENGTH:
             self.length = value
-        elif isinstance(value, (int, long)):
+        elif isinstance(value, integer_types):
             self.length += value
         else:
             self.length += len(value)
@@ -349,7 +347,7 @@ def _sorted_by_type(fields):
     """
     def key(p):
         key, val = p
-        if isinstance(val, (bytes, unicode)):
+        if isinstance(val, (bytes, text_type)):
             return (0, key)
         else:
             return (1, key)

src/treq/test/test_multipart.py

@@ -10,16 +10,16 @@
 from twisted.trial import unittest
 from zope.interface.verify import verifyObject
 
-from twisted.python import compat
+from six import PY3, text_type
+
 from twisted.internet import task
 from twisted.web.client import FileBodyProducer
 from twisted.web.iweb import UNKNOWN_LENGTH, IBodyProducer
 
 from treq.multipart import MultiPartProducer, _LengthConsumer
 
-if compat._PY3:
+if PY3:
     long = int
-    unicode = compat.unicode
 
 
 class MultiPartProducerTestCase(unittest.TestCase):
@@ -65,7 +65,7 @@ def getOutput(self, producer, with_producer=False):
 
     def newLines(self, value):
 
-        if isinstance(value, unicode):
+        if isinstance(value, text_type):
             return value.replace(u"\n", u"\r\n")
         else:
             return value.replace(b"\n", b"\r\n")

src/treq/test/test_testing.py

@@ -6,14 +6,13 @@
 
 from mock import ANY
 
-from six import text_type, binary_type
+from six import text_type, binary_type, PY3
 
 from twisted.trial.unittest import TestCase
 from twisted.web.client import ResponseFailed
 from twisted.web.error import SchemeNotSupported
 from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
-from twisted.python.compat import _PY3
 
 import treq
 
@@ -307,7 +306,7 @@ def test_repr(self):
         """
         :obj:`HasHeaders` returns a nice string repr.
         """
-        if _PY3:
+        if PY3:
             reprOutput = "HasHeaders({b'a': [b'b']})"
         else:
             reprOutput = "HasHeaders({'a': ['b']})"