Hi mates, I'm new here so I have some elemental doubts, it's possible to config a resource input parameter inside an "op"? I need to inject config_schema={"file_name": str} inside the resource.

@op(config_schema={"file_name": str}, required_resource_keys={"wes_manager"})
def get_dataframe(context):
    context.resource.wes_manager # Do something to configure it


@resource(config_schema={"base_dir": Field(String), "file_name": Field(String)})
def wes_manager_resource(context):
    return WESManager(context.resource_config["base_dir"], context.resource_config["file_name"])

Hi Alvaro, It is possible to modify a resource inside of an op under certain select circumstances, but not recommended. This line of yours referencs a

WESManager

instance:

context.resource.wes_manager # Do something to configure it

If you control this class, you can do anything you want to it inside the op. However, your mutations will only carry over if you are using the in-process executor. For most dagster use cases, you are going to be using multiprocess execution, and in that case resources are constructed separately for each op. So any changes you make to a resource inside one op will not be carried over to the others.

Hi Sean, I understand. My use case is a sensor who detects new files inside a folder and runs a job every time he founds a new file. So the only variable between different jobs is the file name. The WESManager is the responsible for open the file and return a data-set based on it. This is the snippet of the job with the injected WESManager resource:

@job(
    resource_defs={
            "wes_manager": wes_manager_resource.configured({"file_name": "I NEED TO INJECT HERE THE FILE NAME"}),
    },
)

Is possible to inject there the provided file name by the sensor who fires the job? Or should be a parameter inside the resource called function? Are resources intended to use only static attributes in their configuration? Thanks for take time to respond me, really appreciate it.

If I understand your problem correctly, you can solve it using runtime resource configuration:

@resource(config_schema={"base_dir": Field(String), "file_name": Field(String)})
def wes_manager(context):
    return WESManager(context.resource_config["base_dir"], context.resource_config["file_name"])


@job(
    resource_defs={
            "wes_manager": wes_manager_resource
    },
)
def some_job():
    # ...

@sensor(job=some_job)
def my_sensor():
    # ... some code ...
    return RunRequest(
        run_config={"resources": {"wes_manager": {"config": {"base_dir": "the_base_dir", "file_name": "the_file_name"}}}}
    )

Oh thanks, that exactly what I need. Really appreciate 🎉