[dagster-pipes] add ThreadedPipesMessageReader with extra runtime params

dagster-io · Oct 2, 2024 · 681425a · 681425a
1 parent 4521b39
commit 681425a
Show file tree

Hide file tree

Showing 3 changed files with 215 additions and 66 deletions.
diff --git a/python_modules/dagster/dagster/_core/pipes/context.py b/python_modules/dagster/dagster/_core/pipes/context.py
@@ -161,7 +161,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":