Put dependency-free websockets protocol in websockets_protocol.py

Author: leszekhanusz

gql/transport/appsync_websockets.py

@@ -224,18 +224,6 @@ async def execute(
     )
     _wait_ack = WebsocketsTransport._wait_ack
 
-    @property
-    def url(self) -> str:
-        return self.adapter.url
-
-    @property
-    def headers(self) -> Dict[str, str]:
-        return self.adapter.headers
-
     @property
     def ssl(self) -> Union[SSLContext, bool]:
         return self.adapter.ssl
-
-    @property
-    def connect_args(self) -> Dict[str, Any]:
-        return self.adapter.connect_args

gql/transport/common/adapters/connection.py

@@ -1,5 +1,5 @@
 import abc
-from typing import Dict
+from typing import Any, Dict
 
 
 class AdapterConnection(abc.ABC):
@@ -8,6 +8,9 @@ class AdapterConnection(abc.ABC):
     This allows different WebSocket implementations to be used interchangeably.
     """
 
+    url: str
+    connect_args: Dict[str, Any]
+
     @abc.abstractmethod
     async def connect(self) -> None:
         """Connect to the server."""

gql/transport/common/adapters/websockets.py

@@ -19,7 +19,7 @@ def __init__(
         *,
         headers: Optional[HeadersLike] = None,
         ssl: Union[SSLContext, bool] = False,
-        connect_args: Dict[str, Any] = {},
+        connect_args: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Initialize the transport with the given parameters.
 
@@ -31,6 +31,10 @@ def __init__(
         self.url: str = url
         self._headers: Optional[HeadersLike] = headers
         self.ssl: Union[SSLContext, bool] = ssl
+
+        if connect_args is None:
+            connect_args = {}
+
         self.connect_args = connect_args
 
         self.websocket: Optional[WebSocketClientProtocol] = None

gql/transport/common/base.py

@@ -559,3 +559,11 @@ async def wait_closed(self) -> None:
             log.debug("Timer close_timeout fired in wait_closed")
 
         log.debug("wait_close: done")
+
+    @property
+    def url(self) -> str:
+        return self.adapter.url
+
+    @property
+    def connect_args(self) -> Dict[str, Any]:
+        return self.adapter.connect_args