summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorBert JW Regeer <bertjw@regeer.org>2020-05-07 00:45:08 -0700
committerBert JW Regeer <bertjw@regeer.org>2020-08-15 18:55:05 -0700
commitb4a0094a2cf21fa76f9d48eb92303dd758c967e2 (patch)
tree6ba73596c9d4a83575e6c8348a95b93656b60f2e
parent9ceda5383da82f8118e19753ddf9ed6c631046fc (diff)
downloadwaitress-b4a0094a2cf21fa76f9d48eb92303dd758c967e2.tar.gz
Cleanup waitress.compat
-rw-r--r--src/waitress/compat.py163
1 files changed, 25 insertions, 138 deletions
diff --git a/src/waitress/compat.py b/src/waitress/compat.py
index fe72a76..aaf7a78 100644
--- a/src/waitress/compat.py
+++ b/src/waitress/compat.py
@@ -1,135 +1,52 @@
import os
-import sys
-import types
import platform
-import warnings
-
-try:
- import urlparse
-except ImportError: # pragma: no cover
- from urllib import parse as urlparse
-try:
- import fcntl
-except ImportError: # pragma: no cover
- fcntl = None # windows
+# Fix for issue reported in https://github.com/Pylons/waitress/issues/138,
+# Python on Windows may not define IPPROTO_IPV6 in socket.
+import socket
+import sys
+import warnings
+from http import client as httplib
+from io import StringIO as NativeIO
+from urllib import parse as urlparse
+from urllib.parse import unquote_to_bytes
-# True if we are running on Python 3.
-PY2 = sys.version_info[0] == 2
-PY3 = sys.version_info[0] == 3
+import _thread as thread
# True if we are running on Windows
WIN = platform.system() == "Windows"
-if PY3: # pragma: no cover
- string_types = (str,)
- integer_types = (int,)
- class_types = (type,)
- text_type = str
- binary_type = bytes
- long = int
-else:
- string_types = (basestring,)
- integer_types = (int, long)
- class_types = (type, types.ClassType)
- text_type = unicode
- binary_type = str
- long = long
-
-if PY3: # pragma: no cover
- from urllib.parse import unquote_to_bytes
-
- def unquote_bytes_to_wsgi(bytestring):
- return unquote_to_bytes(bytestring).decode("latin-1")
+string_types = (str,)
+integer_types = (int,)
+class_types = (type,)
+text_type = str
+binary_type = bytes
+long = int
-else:
- from urlparse import unquote as unquote_to_bytes
-
- def unquote_bytes_to_wsgi(bytestring):
- return unquote_to_bytes(bytestring)
+def unquote_bytes_to_wsgi(bytestring):
+ return unquote_to_bytes(bytestring).decode("latin-1")
def text_(s, encoding="latin-1", errors="strict"):
""" If ``s`` is an instance of ``binary_type``, return
``s.decode(encoding, errors)``, otherwise return ``s``"""
+
if isinstance(s, binary_type):
return s.decode(encoding, errors)
- return s # pragma: no cover
-
-
-if PY3: # pragma: no cover
-
- def tostr(s):
- if isinstance(s, text_type):
- s = s.encode("latin-1")
- return str(s, "latin-1", "strict")
-
- def tobytes(s):
- return bytes(s, "latin-1")
-
-
-else:
- tostr = str
- def tobytes(s):
- return s
-
-
-if PY3: # pragma: no cover
- import builtins
-
- exec_ = getattr(builtins, "exec")
+ return s # pragma: no cover
- def reraise(tp, value, tb=None):
- if value is None:
- value = tp
- if value.__traceback__ is not tb:
- raise value.with_traceback(tb)
- raise value
- del builtins
+def tostr(s):
+ return str(s, "latin-1", "strict")
-else: # pragma: no cover
- def exec_(code, globs=None, locs=None):
- """Execute code in a namespace."""
- if globs is None:
- frame = sys._getframe(1)
- globs = frame.f_globals
- if locs is None:
- locs = frame.f_locals
- del frame
- elif locs is None:
- locs = globs
- exec("""exec code in globs, locs""")
-
- exec_(
- """def reraise(tp, value, tb=None):
- raise tp, value, tb
-"""
- )
-
-try:
- from StringIO import StringIO as NativeIO
-except ImportError: # pragma: no cover
- from io import StringIO as NativeIO
-
-try:
- import httplib
-except ImportError: # pragma: no cover
- from http import client as httplib
-
-try:
- MAXINT = sys.maxint
-except AttributeError: # pragma: no cover
- MAXINT = sys.maxsize
+def tobytes(s):
+ return bytes(s, "latin-1")
-# Fix for issue reported in https://github.com/Pylons/waitress/issues/138,
-# Python on Windows may not define IPPROTO_IPV6 in socket.
-import socket
-
+MAXINT = sys.maxsize
HAS_IPV6 = socket.has_ipv6
if hasattr(socket, "IPPROTO_IPV6") and hasattr(socket, "IPV6_V6ONLY"):
@@ -147,33 +64,3 @@ else: # pragma: no cover
RuntimeWarning,
)
HAS_IPV6 = False
-
-
-def set_nonblocking(fd): # pragma: no cover
- if PY3 and sys.version_info[1] >= 5:
- os.set_blocking(fd, False)
- elif fcntl is None:
- raise RuntimeError("no fcntl module present")
- else:
- flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
- flags = flags | os.O_NONBLOCK
- fcntl.fcntl(fd, fcntl.F_SETFL, flags)
-
-
-if PY3:
- ResourceWarning = ResourceWarning
-else:
- ResourceWarning = UserWarning
-
-
-def qualname(cls):
- if PY3:
- return cls.__qualname__
- return cls.__name__
-
-
-try:
- import thread
-except ImportError:
- # py3
- import _thread as thread