Hello, I am not sure if I am using the configured decorator correctly in this case and need some help. Using

dagster

version

1.16

and

dagster-dbt

version

0.17.6

. When passing the target-path into the configured decorator of the

dbt_cli_resource

,

context.resources.dbt.run

executes correctly and compiles the code in the specific target path, but dagster fails when getting the results into the

DbtCliOutput

because it's still looking for

run_results.json

in the default target path.. This is my configuration of the dbt_cli_resource: get_dbt_vars resturns a simple structure e.g. '{"customer_code": "abc"}'

@configured(dbt_cli_resource, config_schema={"customer_code": str})
def customer_dbt_cli_resource(config):
    vars = get_dbt_vars_for_customer(config["customer_code"])
    return {
        "vars": vars,
        "project-dir": constants.DBT_PROJECT_PATH,
        "profiles-dir": constants.DBT_PROFILES,
        "target-path": f"target/{config['customer_code']}/"
    }

This is my custom dbt run op, which collects only the models I need to run for a particular customer:

@op(required_resource_keys={"dbt"})
def op_dbt_run(context, run_config):

    # is there a way to use dagster-dbt op dbt_run_op and select the models to run and pass in variables?
    models_to_run = run_config.get_models_to_run()
    dbt_output = context.resources.dbt.run(select=models_to_run, vars=get_dbt_vars_for_customer(run_config.customer_code))
    return run_config

This is the job I am using to test:

@job(resource_defs={
            "run_config": make_values_resource(),
            "dbt": customer_dbt_cli_resource
        })
def job_dbt():
        op_dbt_run(op_intantiate_run_config())

When running the job, this is the stack trace:

dagster_dbt.errors.DagsterDbtCliOutputsNotFoundError: Expected to find file at path C:\Users\manan patel\OneDrive - Altis Consulting P L\Documents\Projects\Clients\2022\Mutinex\repo\monorepo\dagster\data_pipelines\../../dbt/data_pipelines/target\run_results.json
  File "C:\Python310\lib\site-packages\dagster\_core\execution\plan\execute_plan.py", line 266, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "C:\Python310\lib\site-packages\dagster\_core\execution\plan\execute_step.py", line 388, in core_dagster_event_sequence_for_step
    for user_event in check.generator(
  File "C:\Python310\lib\site-packages\dagster\_core\execution\plan\execute_step.py", line 94, in _step_output_error_checked_user_event_sequence
    for user_event in user_event_sequence:
  File "C:\Python310\lib\site-packages\dagster\_core\execution\plan\compute.py", line 177, in execute_core_compute
    for step_output in _yield_compute_results(step_context, inputs, compute_fn):
  File "C:\Python310\lib\site-packages\dagster\_core\execution\plan\compute.py", line 145, in _yield_compute_results
    for event in iterate_with_context(
  File "C:\Python310\lib\site-packages\dagster\_utils\__init__.py", line 458, in iterate_with_context
    next_output = next(iterator)
  File "C:\Python310\lib\site-packages\dagster\_core\execution\plan\compute_generator.py", line 75, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "C:\Users\manan patel\OneDrive - Altis Consulting P L\Documents\Projects\Clients\2022\Mutinex\repo\monorepo\dagster\data_pipelines\data_pipelines.py", line 109, in op_dbt_run
    dbt_output = context.resources.dbt.run(select=models_to_run, vars=get_dbt_vars_for_customer(run_config.customer_code))
  File "C:\Python310\lib\site-packages\dagster_dbt\cli\resources.py", line 145, in run
    return self.cli("run", models=models, exclude=exclude, select=select, **kwargs)
  File "C:\Python310\lib\site-packages\dagster_dbt\cli\resources.py", line 90, in cli
    return execute_cli(
  File "C:\Python310\lib\site-packages\dagster_dbt\cli\utils.py", line 145, in execute_cli
    parse_run_results(flags_dict["project-dir"], target_path)
  File "C:\Python310\lib\site-packages\dagster_dbt\cli\utils.py", line 167, in parse_run_results
    raise DagsterDbtCliOutputsNotFoundError(path=run_results_path)

hi @Manan P! sorry that this error message wasn't particularly helpful, as this is a pretty sneaky error. I believe if you instead do

return {
        "vars": vars,
        "project-dir": constants.DBT_PROJECT_PATH,
        "profiles-dir": constants.DBT_PROFILES,
        "target_path": f"target/{config['customer_code']}/"
    }

then things should work for you (i.e., replacing

target-path

with

target_path

)