Migrate from dagster-shell to PipesSubprocessClient

dagster_meltano/meltano_resource.py

@@ -6,13 +6,12 @@
 from pathlib import Path
 from typing import Dict, List, Optional, Union
 
-from dagster import DagsterLogManager, resource, Field
+from dagster import DagsterLogManager, resource, Field, OpExecutionContext, PipesSubprocessClient
 from dagster_meltano.exceptions import MeltanoCommandError
 
 from dagster_meltano.job import Job
 from dagster_meltano.schedule import Schedule
 from dagster_meltano.utils import Singleton
-from dagster_shell import execute_shell_command
 
 STDOUT = 1
 
@@ -46,32 +45,40 @@ def execute_command(
         self,
         command: str,
         env: Dict[str, str],
-        logger: Union[logging.Logger, DagsterLogManager] = logging.Logger,
+        context: OpExecutionContext,
     ) -> str:
         """Execute a Meltano command.
 
         Args:
-            context (OpExecutionContext): The Dagster execution context.
             command (str): The Meltano command to execute.
             env (Dict[str, str]): The environment variables to inject when executing the command.
+            context (OpExecutionContext): The Dagster execution context.
 
         Returns:
             str: The output of the command.
         """
-        output, exit_code = execute_shell_command(
-            f"{self.meltano_bin} {command}",
-            env={**self.default_env, **env},
-            output_logging="STREAM",
-            log=logger,
+        full_command = f"{self.meltano_bin} {command}"
+        merged_env = {**self.default_env, **env}
+
+        context.log.info(f"Executing command with PipesSubprocessClient: {full_command}")
+
+        client: PipesSubprocessClient = PipesSubprocessClient(
+            env=merged_env,
             cwd=self.project_dir,
+            context_injector=None,
         )
+        _command: list[str] = full_command.split(" ")
+        client.run(
+            command=_command,
+            context=context,
+        )
+
+        # if exit_code != 0:
+        #     raise MeltanoCommandError(
+        #         f"Command '{command}' failed with exit code {exit_code}"
+        #     )
 
-        if exit_code != 0:
-            raise MeltanoCommandError(
-                f"Command '{command}' failed with exit code {exit_code}"
-            )
-
-        return output
+        return ""
 
     async def load_json_from_cli(self, command: List[str]) -> dict:
         """Use the Meltano CLI to load JSON data.

dagster_meltano/ops.py

@@ -90,7 +90,7 @@ def dagster_op(
         env = {**env, **config_env}
 
         # Run the Meltano command
-        output = meltano_resource.execute_command(f"{command}", env, context.log)
+        output = meltano_resource.execute_command(f"{command}", env, context)
 
         # Return the logs
         return output