Broken typing annotations fixed (#949)

* Broken typing annotations fixed

* Annotations converted from Python 3.10 format to Python 3.8 format

* Replace | with Union[] in _utils.py

---------

Co-authored-by: Vasily Zakharov <v.zakharov@wwpass.com>
Co-authored-by: engn33r <engn33r@users.noreply.github.com>

websocket/_abnf.py

@@ -3,9 +3,11 @@
 import struct
 import sys
 
+from threading import Lock
+from typing import Callable, Union
+
 from ._exceptions import *
 from ._utils import validate_utf8
-from threading import Lock
 
 """
 _abnf.py
@@ -42,9 +44,9 @@ def _mask(_m, _d) -> bytes:
 
     def _mask(mask_value: array.array, data_value: array.array) -> bytes:
         datalen = len(data_value)
-        data_value = int.from_bytes(data_value, native_byteorder)
-        mask_value = int.from_bytes(mask_value * (datalen // 4) + mask_value[: datalen % 4], native_byteorder)
-        return (data_value ^ mask_value).to_bytes(datalen, native_byteorder)
+        int_data_value = int.from_bytes(data_value, native_byteorder)
+        int_mask_value = int.from_bytes(mask_value * (datalen // 4) + mask_value[: datalen % 4], native_byteorder)
+        return (int_data_value ^ int_mask_value).to_bytes(datalen, native_byteorder)
 
 
 __all__ = [
@@ -132,7 +134,7 @@ class ABNF:
     LENGTH_63 = 1 << 63
 
     def __init__(self, fin: int = 0, rsv1: int = 0, rsv2: int = 0, rsv3: int = 0,
-                 opcode: int = OPCODE_TEXT, mask: int = 1, data: str or bytes = "") -> None:
+                 opcode: int = OPCODE_TEXT, mask: int = 1, data: Union[str, bytes] = "") -> None:
         """
         Constructor for ABNF. Please check RFC for arguments.
         """
@@ -187,7 +189,7 @@ def __str__(self) -> str:
             + " data=" + str(self.data)
 
     @staticmethod
-    def create_frame(data: str, opcode: int, fin: int = 1) -> 'ABNF':
+    def create_frame(data: Union[bytes, str], opcode: int, fin: int = 1) -> 'ABNF':
         """
         Create frame to send text, binary and other data.
 
@@ -237,7 +239,7 @@ def format(self) -> bytes:
             mask_key = self.get_mask_key(4)
             return frame_header + self._get_masked(mask_key)
 
-    def _get_masked(self, mask_key: str or bytes) -> bytes:
+    def _get_masked(self, mask_key: Union[str, bytes]) -> bytes:
         s = ABNF.mask(mask_key, self.data)
 
         if isinstance(mask_key, str):
@@ -246,7 +248,7 @@ def _get_masked(self, mask_key: str or bytes) -> bytes:
         return mask_key + s
 
     @staticmethod
-    def mask(mask_key: str or bytes, data: str or bytes) -> bytes:
+    def mask(mask_key: Union[str, bytes], data: Union[str, bytes]) -> bytes:
         """
         Mask or unmask data. Just do xor for each byte
 
@@ -273,7 +275,7 @@ class frame_buffer:
     _HEADER_MASK_INDEX = 5
     _HEADER_LENGTH_INDEX = 6
 
-    def __init__(self, recv_fn: int, skip_utf8_validation: bool) -> None:
+    def __init__(self, recv_fn: Callable[[int], int], skip_utf8_validation: bool) -> None:
         self.recv = recv_fn
         self.skip_utf8_validation = skip_utf8_validation
         # Buffers over the packets from the layer beneath until desired amount
@@ -304,7 +306,7 @@ def recv_header(self) -> None:
 
         self.header = (fin, rsv1, rsv2, rsv3, opcode, has_mask, length_bits)
 
-    def has_mask(self) -> bool or int:
+    def has_mask(self) -> Union[bool, int]:
         if not self.header:
             return False
         return self.header[frame_buffer._HEADER_MASK_INDEX]
@@ -410,7 +412,7 @@ def add(self, frame: ABNF) -> None:
         if frame.fin:
             self.recving_frames = None
 
-    def is_fire(self, frame: ABNF) -> bool or int:
+    def is_fire(self, frame: ABNF) -> Union[bool, int]:
         return frame.fin or self.fire_cont_frame
 
     def extract(self, frame: ABNF) -> list:

websocket/_app.py

@@ -1,12 +1,12 @@
 import inspect
 import selectors
+import socket
 import sys
 import threading
 import time
 import traceback
-import socket
 
-from typing import Callable, Any
+from typing import Any, Callable, Optional, Union
 
 from . import _logging
 from ._abnf import ABNF
@@ -139,7 +139,7 @@ class WebSocketApp:
     Higher level of APIs are provided. The interface is like JavaScript WebSocket object.
     """
 
-    def __init__(self, url: str, header: list or dict or Callable = None,
+    def __init__(self, url: str, header: Union[list, dict, Callable] = None,
                  on_open: Callable = None, on_message: Callable = None, on_error: Callable = None,
                  on_close: Callable = None, on_ping: Callable = None, on_pong: Callable = None,
                  on_cont_message: Callable = None,
@@ -289,9 +289,9 @@ def _send_ping(self) -> None:
                     _logging.debug("Failed to send ping: {err}".format(err=e))
 
     def run_forever(self, sockopt: tuple = None, sslopt: dict = None,
-                    ping_interval: float = 0, ping_timeout: float or None = None,
+                    ping_interval: float = 0, ping_timeout: Optional[float] = None,
                     ping_payload: str = "",
-                    http_proxy_host: str = None, http_proxy_port: int or str = None,
+                    http_proxy_host: str = None, http_proxy_port: Union[int, str] = None,
                     http_no_proxy: list = None, http_proxy_auth: tuple = None,
                     http_proxy_timeout: float = None,
                     skip_utf8_validation: bool = False,

websocket/_cookiejar.py

@@ -1,5 +1,7 @@
 import http.cookies
 
+from typing import Optional
+
 """
 _cookiejar.py
 websocket - WebSocket client library for Python
@@ -24,7 +26,7 @@ class SimpleCookieJar:
     def __init__(self) -> None:
         self.jar = dict()
 
-    def add(self, set_cookie: str) -> None:
+    def add(self, set_cookie: Optional[str]) -> None:
         if set_cookie:
             simpleCookie = http.cookies.SimpleCookie(set_cookie)
 

websocket/_core.py

@@ -3,6 +3,8 @@
 import threading
 import time
 
+from typing import Optional, Union
+
 # websocket modules
 from ._abnf import *
 from ._exceptions import *
@@ -144,7 +146,7 @@ def gettimeout(self) -> float:
         """
         return self.sock_opt.timeout
 
-    def settimeout(self, timeout: float):
+    def settimeout(self, timeout: Optional[float]):
         """
         Set the timeout to the websocket.
 
@@ -265,7 +267,7 @@ def connect(self, url, **options):
                 self.sock = None
             raise
 
-    def send(self, payload: bytes or str, opcode: int = ABNF.OPCODE_TEXT) -> int:
+    def send(self, payload: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> int:
         """
         Send the data as string.
 
@@ -324,7 +326,7 @@ def send_binary(self, payload: bytes) -> int:
         """
         return self.send(payload, ABNF.OPCODE_BINARY)
 
-    def ping(self, payload: str or bytes = ""):
+    def ping(self, payload: Union[str, bytes] = ""):
         """
         Send ping data.
 
@@ -337,7 +339,7 @@ def ping(self, payload: str or bytes = ""):
             payload = payload.encode("utf-8")
         self.send(payload, ABNF.OPCODE_PING)
 
-    def pong(self, payload: str or bytes = ""):
+    def pong(self, payload: Union[str, bytes] = ""):
         """
         Send pong data.
 
@@ -350,7 +352,7 @@ def pong(self, payload: str or bytes = ""):
             payload = payload.encode("utf-8")
         self.send(payload, ABNF.OPCODE_PONG)
 
-    def recv(self) -> str or bytes:
+    def recv(self) -> Union[str, bytes]:
         """
         Receive string data(byte array) from the server.
 
@@ -521,7 +523,7 @@ def shutdown(self):
             self.sock = None
             self.connected = False
 
-    def _send(self, data: str or bytes):
+    def _send(self, data: Union[str, bytes]):
         return send(self.sock, data)
 
     def _recv(self, bufsize):

websocket/_socket.py

@@ -2,6 +2,8 @@
 import selectors
 import socket
 
+from typing import Union
+
 from ._exceptions import *
 from ._ssl_compat import *
 from ._utils import *
@@ -53,7 +55,7 @@ def __init__(self, sockopt: list, sslopt: dict) -> None:
         self.timeout = None
 
 
-def setdefaulttimeout(timeout: int or float) -> None:
+def setdefaulttimeout(timeout: Union[int, float, None]) -> None:
     """
     Set the global timeout setting to connect.
 
@@ -66,7 +68,7 @@ def setdefaulttimeout(timeout: int or float) -> None:
     _default_timeout = timeout
 
 
-def getdefaulttimeout() -> int or float:
+def getdefaulttimeout() -> Union[int, float, None]:
     """
     Get default timeout
 
@@ -135,7 +137,7 @@ def recv_line(sock: socket.socket) -> bytes:
     return b''.join(line)
 
 
-def send(sock: socket.socket, data: bytes) -> int:
+def send(sock: socket.socket, data: Union[bytes, str]) -> int:
     if isinstance(data, str):
         data = data.encode('utf-8')
 

websocket/_url.py

@@ -2,6 +2,7 @@
 import socket
 import struct
 
+from typing import Optional
 from urllib.parse import unquote, urlparse
 
 """
@@ -101,7 +102,7 @@ def _is_address_in_network(ip: str, net: str) -> bool:
     return ipaddr & netmask == netaddr
 
 
-def _is_no_proxy_host(hostname: str, no_proxy: list) -> bool:
+def _is_no_proxy_host(hostname: str, no_proxy: Optional[list]) -> bool:
     if not no_proxy:
         v = os.environ.get("no_proxy", os.environ.get("NO_PROXY", "")).replace(" ", "")
         if v:
@@ -122,8 +123,8 @@ def _is_no_proxy_host(hostname: str, no_proxy: list) -> bool:
 
 
 def get_proxy_info(
-        hostname: str, is_secure: bool, proxy_host: str = None, proxy_port: int = 0, proxy_auth: tuple = None,
-        no_proxy: list = None, proxy_type: str = 'http') -> tuple:
+        hostname: str, is_secure: bool, proxy_host: Optional[str] = None, proxy_port: int = 0, proxy_auth: Optional[tuple] = None,
+        no_proxy: Optional[list] = None, proxy_type: str = 'http') -> tuple:
     """
     Try to retrieve proxy host and port from environment
     if not provided in options.

websocket/_utils.py

@@ -1,3 +1,5 @@
+from typing import Union
+
 """
 _url.py
 websocket - WebSocket client library for Python
@@ -72,7 +74,7 @@ def _decode(state: int, codep: int, ch: int) -> tuple:
 
         return state, codep
 
-    def _validate_utf8(utfbytes: str or bytes) -> bool:
+    def _validate_utf8(utfbytes: Union[str, bytes]) -> bool:
         state = _UTF8_ACCEPT
         codep = 0
         for i in utfbytes:
@@ -83,7 +85,7 @@ def _validate_utf8(utfbytes: str or bytes) -> bool:
         return True
 
 
-def validate_utf8(utfbytes: str or bytes) -> bool:
+def validate_utf8(utfbytes: Union[str, bytes]) -> bool:
     """
     validate utf8 byte string.
     utfbytes: utf byte string to check.
@@ -92,13 +94,13 @@ def validate_utf8(utfbytes: str or bytes) -> bool:
     return _validate_utf8(utfbytes)
 
 
-def extract_err_message(exception: Exception) -> str or None:
+def extract_err_message(exception: Exception) -> Union[str, None]:
     if exception.args:
         return exception.args[0]
     else:
         return None
 
 
-def extract_error_code(exception: Exception) -> int or None:
+def extract_error_code(exception: Exception) -> Union[int, None]:
     if exception.args and len(exception.args) > 1:
         return exception.args[0] if isinstance(exception.args[0], int) else None