Remove PY2.

opensearchpy/client/utils.py

@@ -32,7 +32,7 @@
 from datetime import date, datetime
 from functools import wraps
 
-from ..compat import PY2, quote, string_types, to_bytes, to_str, unquote, urlparse
+from ..compat import quote, string_types, to_bytes, to_str, unquote, urlparse
 
 # parts of URL to be omitted
 SKIP_IN_PATH = (None, "", b"", [], ())
@@ -107,9 +107,7 @@ def _escape(value):
 
     # encode strings to utf-8
     if isinstance(value, string_types):
-        if PY2 and isinstance(value, unicode):  # noqa: F821
-            return value.encode("utf-8")
-        if not PY2 and isinstance(value, str):
+        if isinstance(value, str):
             return value.encode("utf-8")
 
     return str(value)

opensearchpy/compat.py

@@ -25,41 +25,23 @@
 #  under the License.
 
 
-import sys
+from queue import Queue
+from urllib.parse import quote, quote_plus, unquote, urlencode, urlparse
 
-PY2 = sys.version_info[0] == 2
+string_types = str, bytes
+map = map
 
-if PY2:
-    string_types = (basestring,)  # noqa: F821
-    from itertools import imap as map
-    from urllib import quote, quote_plus, unquote, urlencode
 
-    from Queue import Queue
-    from urlparse import urlparse
+def to_str(x, encoding="ascii"):
+    if not isinstance(x, str):
+        return x.decode(encoding)
+    return x
 
-    def to_str(x, encoding="ascii"):
-        if not isinstance(x, str):
-            return x.encode(encoding)
-        return x
 
-    to_bytes = to_str
-
-else:
-    string_types = str, bytes
-    from urllib.parse import quote, quote_plus, unquote, urlencode, urlparse
-
-    map = map
-    from queue import Queue
-
-    def to_str(x, encoding="ascii"):
-        if not isinstance(x, str):
-            return x.decode(encoding)
-        return x
-
-    def to_bytes(x, encoding="ascii"):
-        if not isinstance(x, bytes):
-            return x.encode(encoding)
-        return x
+def to_bytes(x, encoding="ascii"):
+    if not isinstance(x, bytes):
+        return x.encode(encoding)
+    return x
 
 
 try:

opensearchpy/compat.pyi

@@ -27,7 +27,6 @@
 import sys
 from typing import Callable, Tuple, Type, Union
 
-PY2: bool
 string_types: Tuple[type, ...]
 
 to_str: Callable[[Union[str, bytes]], str]

test_opensearchpy/test_client/test_utils.py

@@ -29,9 +29,8 @@
 from __future__ import unicode_literals
 
 from opensearchpy.client.utils import _bulk_body, _escape, _make_path, query_params
-from opensearchpy.compat import PY2
 
-from ..test_cases import SkipTest, TestCase
+from ..test_cases import TestCase
 
 
 class TestQueryParams(TestCase):
@@ -161,14 +160,6 @@ def test_handles_unicode(self):
             "/some-index/type/%E4%B8%AD%E6%96%87", _make_path("some-index", "type", id)
         )
 
-    def test_handles_utf_encoded_string(self):
-        if not PY2:
-            raise SkipTest("Only relevant for py2")
-        id = "中文".encode("utf-8")
-        self.assertEqual(
-            "/some-index/type/%E4%B8%AD%E6%96%87", _make_path("some-index", "type", id)
-        )
-
 
 class TestEscape(TestCase):
     def test_handles_ascii(self):

test_opensearchpy/test_connection/test_base_connection.py

@@ -28,13 +28,9 @@
 
 import os
 import sys
-import unittest
 import warnings
 
-import six
-
 from opensearchpy.connection import Connection
-from opensearchpy.exceptions import NotFoundError
 
 from ..test_cases import TestCase
 
@@ -92,19 +88,6 @@ def test_raises_warnings_when_folded(self):
 
         self.assertEqual([str(w.message) for w in warn], ["warning", "folded"])
 
-    @unittest.skipIf(six.PY2, "not compatible with python2")
-    def test_raises_errors(self):
-        con = Connection()
-        with self.assertLogs("opensearch") as captured, self.assertRaises(
-            NotFoundError
-        ):
-            con._raise_error(404, "Not found", "application/json")
-        self.assertEqual(len(captured.output), 1)
-
-        # NB: this should assertNoLogs() but that method is not available until python3.10
-        with self.assertRaises(NotFoundError):
-            con._raise_error(404, "Not found", "text/plain; charset=UTF-8")
-
     def test_ipv6_host_and_port(self):
         for kwargs, expected_host in [
             ({"host": "::1"}, "http://[::1]:9200"),