Zach
08/18/2023, 6:47 PMfrom time import sleep
from dagster import ConfigurableResource, StepLauncher, op, job
class NewStepLauncher(StepLauncher, ConfigurableResource):
def launch_step(self, step_context):
<http://step_context.log.info|step_context.log.info>("Launching step from custom StepLauncher")
sleep(10)
@op()
def my_custom_step(context, step_launcher: NewStepLauncher):
pass
@job(resource_defs={"step_launcher": NewStepLauncher})
def my_custom_job():
my_custom_step()
defs = Definitions(
resources={"step_launcher": NewStepLauncher().configure_at_launch()},
jobs=[my_custom_job],
)
and while the code loads fine, it doesn't seem to be using the NewStepLauncher.launch_step
method to launch my_custom_step
Zach
08/18/2023, 7:09 PMZach
08/18/2023, 8:01 PMrequired_resource_keys
. Unfortunately the majority of my team's jobs use a step launcher 😿sean
08/18/2023, 10:00 PMZach
08/18/2023, 10:02 PM