[dagster-pipes] add ThreadedPipesMessageReader with extra runtime params

dagster-io · Sep 30, 2024 · 7f7242b · 7f7242b
1 parent a993cd9
commit 7f7242b
Show file tree

Hide file tree

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