As dagster_shell is deprecated, swith to PipesSubprocessClient

pipeline/__init__.py

@@ -2,6 +2,7 @@
     DefaultScheduleStatus,
     Definitions,
     EnvVar,
+    PipesSubprocessClient,
     ScheduleDefinition,
     define_asset_job,
     load_assets_from_modules,
@@ -25,6 +26,7 @@
             password=EnvVar('IPL_POSTGRES_PASSWORD'),
             database=EnvVar('IPL_POSTGRES_DB'),
         ),
+        'pipes_subprocess_client': PipesSubprocessClient(),
         'json_webasset_io_manager': JsonWebAssetIOManager(
             destination_directory=EnvVar('WWW_ROOT_DIR'),
         ),

pipeline/assets/webcams.py

@@ -1,15 +1,17 @@
 import os
 import random
 import warnings
+from typing import Sequence
 
 from dagster import (
     AssetExecutionContext,
     AutoMaterializePolicy,
     ExperimentalWarning,
     FreshnessPolicy,
+    PipesExecutionResult,
+    PipesSubprocessClient,
     asset,
 )
-from dagster_shell import execute_shell_command
 
 SCRIPT_DIR = os.getenv('SCRIPT_DIR', './scripts/')
 
@@ -20,13 +22,13 @@
     freshness_policy=FreshnessPolicy(maximum_lag_minutes=2, cron_schedule='* * * * *'),
     auto_materialize_policy=AutoMaterializePolicy.eager(),
 )
-def webcam_images(context: AssetExecutionContext) -> None:
+def webcam_images(
+    context: AssetExecutionContext, pipes_subprocess_client: PipesSubprocessClient
+) -> Sequence['PipesExecutionResult']:
     """
     Downloads webcam images via lftp.
     """
-    (logs, exit_code) = execute_shell_command(
-        'bash download_webcams.sh', cwd=SCRIPT_DIR, output_logging='STREAM', log=context.log
-    )
 
-    if exit_code != 0:
-        raise RuntimeError(f'Downloading webcam images failed with error code {exit_code}')
+    return pipes_subprocess_client.run(
+        command=['bash', 'download_webcams.sh'], context=context, cwd=SCRIPT_DIR
+    ).get_results()

requirements-dagster.txt

@@ -3,4 +3,3 @@ dagster-graphql==1.7.12
 dagster-webserver==1.7.12
 dagster-postgres==0.23.12
 dagster-docker==0.23.12
-dagster-shell==0.23.12

requirements-pipeline.txt

@@ -1,7 +1,6 @@
 dagster==1.7.12
 dagster-postgres==0.23.12
 dagster-docker==0.23.12
-dagster-shell==0.23.12
 geopandas==0.14.3
 GeoAlchemy2==0.14.4
 # Because a specific version of the PyPi GDAL package depends on specific OS library versions, and because Ubuntu (LTS) currently only provides *older* versions of them, we ping GDAL to v3.6 here.