lint

python/pyspark/sql/connect/client/core.py

@@ -1068,14 +1068,14 @@ def execute_command_as_iterator(
         if self._user_id:
             req.user_context.user_id = self._user_id
         req.plan.command.CopyFrom(command)
-        for response in self._execute_and_fetch_as_iterator(req, observations):
+        for response in self._execute_and_fetch_as_iterator(req, observations or {}):
             if isinstance(response, dict):
                 yield response
             else:
                 raise PySparkValueError(
                     error_class="UNKNOWN_RESPONSE",
                     message_parameters={
-                        "response": response,
+                        "response": str(response),
                     },
                 )
 
@@ -1343,8 +1343,8 @@ def handle_response(
                 cmd_result = b.streaming_query_manager_command_result
                 yield {"streaming_query_manager_command_result": cmd_result}
             if b.HasField("streaming_query_listener_events_result"):
-                cmd_result = b.streaming_query_listener_events_result
-                yield {"streaming_query_listener_events_result": cmd_result}
+                event_result = b.streaming_query_listener_events_result
+                yield {"streaming_query_listener_events_result": event_result}
             if b.HasField("get_resources_command_result"):
                 resources = {}
                 for key, resource in b.get_resources_command_result.resources.items():

python/pyspark/sql/connect/streaming/query.py

@@ -270,13 +270,13 @@ class StreamingQueryListenerBus:
     receive listener events and invoke correct listener call backs.
     """
 
-    def __init__(self, sqm: "StreamingQueryManager"):
+    def __init__(self, sqm: "StreamingQueryManager") -> None:
         self._sqm = sqm
         self._listener_bus: List[StreamingQueryListener] = []
         self._execution_thread: Optional[Thread] = None
         self._lock = Lock()
 
-    def append(self, listener: StreamingQueryListener):
+    def append(self, listener: StreamingQueryListener) -> None:
         """
         Append a listener to the local listener bus. When the added listener is
         the first listener, request the server to create the server side listener
@@ -301,7 +301,7 @@ def append(self, listener: StreamingQueryListener):
                 )
                 self._execution_thread.start()
 
-    def remove(self, listener: StreamingQueryListener):
+    def remove(self, listener: StreamingQueryListener) -> None:
         """
         Remove the listener from the local listener bus.
 
@@ -384,7 +384,7 @@ def _query_event_handler(self, iter: Iterator[Dict[str, Any]]) -> None:
 
     @staticmethod
     def deserialize(
-        event: str,
+        event: pb2.StreamingQueryListenerEvent,
     ) -> Union["QueryProgressEvent", "QueryIdleEvent", "QueryTerminatedEvent"]:
         if event.event_type == proto.StreamingQueryEventType.QUERY_PROGRESS_EVENT:
             return QueryProgressEvent.fromJson(json.loads(event.event_json))