Hello, not sure why

setup_for_execution

doesn't trigger for

ConfigurableResource

class in the following example (V1.3.3). Anything I missed here?:

import mlflow
from dagster import op, ConfigurableResource

class MLflowResource(ConfigurableResource):
    name: str

    def setup_for_execution(self, context) -> None:
        print("setup_for_execution")
        mlflow.start_run()

    def log_metric(self, key, value):
        print("log_metric")
        mlflow.log_metric(key, value)

@op
def mlflow_op(mlflow_resource: MLflowResource):
    mlflow_resource.log_metric("a", 1)

if __name__ == "__main__":
    r = MLflowResource(name="mlflow")
    mlflow_op(r)

Hi Jueming,

setup_for_execution

isn’t being triggered because

MLFlowResource

is not being used in the context of a Dagster run. The

setup_for_execution

hook gets called during an initialization sequence that doesn’t occur when you use direct op invocation. This will call `setup_for_execution`:

@job
def my_job():
    mlflow_op()

defs = Definitions(
    jobs=[my_job],
    resources={"mlflow_resource": MLflowResource(name="mlflow")}
)

defs.get_job_def("my_job").execute_in_process()

Hey @sean sorry for the late reply. I do verify it and it works. Thanks for the help! Just wonder will "dagster_mlflow" package be adjusted for the pythonic way to define resource? (it seems that it's still the legacy way in the doc).