[dagster-pipes] add ThreadedPipesMessageReader with extra runtime params

Author: danielgafni

python_modules/dagster/dagster/_core/pipes/client.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Sequence
+from typing import TYPE_CHECKING, Any, Iterator, List, Mapping, Optional, Sequence, TypedDict
 
 from dagster_pipes import (
     DagsterPipesError,
@@ -143,7 +143,18 @@ def no_messages_debug_text(self) -> str:
         """
 
 
+class PipesLaunchedData(TypedDict):
+    """Payload generated on the Client side after the external process startup
+    containing arbitrary information about the external process.
+    """
+
+    extras: Mapping[str, Any]
+
+
 class PipesMessageReader(ABC):
+    launched_payload: Optional[PipesLaunchedData]
+    opened_payload: Optional[PipesOpenedData]
+
     @abstractmethod
     @contextmanager
     def read_messages(self, handler: "PipesMessageHandler") -> Iterator[PipesParams]:
@@ -170,6 +181,12 @@ def on_opened(self, opened_payload: PipesOpenedData) -> None:
         that can only be obtained from the external process.
         """
 
+    def on_launched(self, params: Mapping[str, Any]) -> None:
+        """Can be called manually to submit extra information about the launched process
+        to the message reader. Populates the `launched_payload` attribute.
+        """
+        self.opened_payload = {"extras": params}
+
     @abstractmethod
     def no_messages_debug_text(self) -> str:
         """A message to be displayed when no messages are received from the external process to aid with

python_modules/dagster/dagster/_core/pipes/context.py

@@ -141,7 +141,9 @@ def _resolve_metadata_value(
     # Type ignores because we currently validate in individual handlers
     def handle_message(self, message: PipesMessage) -> None:
         if self._received_closed_msg:
-            self._context.log.warn(f"[pipes] unexpected message received after closed: `{message}`")
+            self._context.log.warning(
+                f"[pipes] unexpected message received after closed: `{message}`"
+            )
 
         method = cast(Method, message["method"])
         if method == "opened":