How do I define a method using dbt.cli that dynamically executes a dbt command?

[emilee-jurkowski]

We have defined dbt like so:

dbt = DbtCliResource(
    project_dir=foo_project_dir, 
    profile="foo",
    target=foo_target
)

@dbt_assets(
    manifest=dbt_manifest_path,
    dagster_dbt_translator=CustomDagsterDbtTranslator(),
    exclude="resource_type:seed",
)
def my_dbt_assets(context: AssetExecutionContext):
    yield from dbt.cli(["build"], context=context).stream()

and are trying to dynamically trigger a dbt cli run with a sensor

 @op(required_resource_keys={"dbt"}, out=Out(Nothing))
 def dbt_cli_op(context: OpExecutionContext, dbt_command: str, dbt_select: str, dbt_exclude: str=None):
     dbt_cli_cmd = [dbt_command, "--select", dbt_select]
     if dbt_exclude:
         dbt_cli_cmd.extend(["--exclude", dbt_exclude])
     yield from context.resources.dbt.cli(dbt_cli_cmd, manifest=dbt_manifest_path, dagster_dbt_translator=CustomDagsterDbtTranslator).stream()

@job(resource_defs={"dbt": dbt})
def dbt_cli_job():
    dbt_cli_op()

@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    monitored_jobs=[dbt_run_operation_job],
    request_job=dbt_cli_job
)
def dbt_build_sensor(context: RunStatusSensorContext):
    if 'dagster/schedule_name' in context.dagster_run.tags:
        dagster_schedule_name = context.dagster_run.tags['dagster/schedule_name']
        dbt_command = "build"

        if dagster_schedule_name == 'my_schedule_1':
            dbt_select = "tag:daily+"
            dbt_exclude = "tag:manual tag:hourly+"

        elif dagster_schedule_name == 'my_schedule_2':
            dbt_select = "tag:hourly+"
            dbt_exclude = "tag:manual tag:daily"

        elif dagster_schedule_name == 'my_schedule_3':
            dbt_select = "source:my_dbt_source.source_1+"
            dbt_exclude = "tag:manual tag:hourly+ tag:daily"

        yield RunRequest(
            run_config={
                "ops": {
                    "dbt_cli_op": {
                        "inputs": {
                            "dbt_command": dbt_command,
                            "dbt_select": dbt_select,
                            "dbt_exclude": dbt_exclude
                        }
                    }
                }
            },
            tags={
                'triggered_by': f'{dagster_schedule_name}',
                'triggering_run_id': f'{context.dagster_run.run_id}'
            }
        )

The sensor will trigger a run, but end up with errors like the following. It's always the same error, sometimes it just displays a different dbt model name.

dagster._core.errors.DagsterInvariantViolationError: Core compute for op "dbt_cli_op" returned an output "my_dbt_model" that does not exist. The available outputs are ['result']

Stack Trace:
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_plan.py", line 275, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
,  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 476, in core_dagster_event_sequence_for_step
    for user_event in _step_output_error_checked_user_event_sequence(
,  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 167, in _step_output_error_checked_user_event_sequence
    raise DagsterInvariantViolationError(

I have also tried out=Out(Any) and just completely removing the out specification, but neither seem to help. Is there another way in which to define the output so that this does not error?

[rexledesma]

You should try setting out={}.

- @op(required_resource_keys={"dbt"}, out=Out(Nothing))
+ @op(required_resource_keys={"dbt"}, out={})
def dbt_cli_op(context: OpExecutionContext, dbt_command: str, dbt_select: str, dbt_exclude: str=None):
    dbt_cli_cmd = [dbt_command, "--select", dbt_select]
    if dbt_exclude:
        dbt_cli_cmd.extend(["--exclude", dbt_exclude])
    yield from context.resources.dbt.cli(dbt_cli_cmd, manifest=dbt_manifest_path, dagster_dbt_translator=CustomDagsterDbtTranslator).stream()

[emilee-jurkowski]

Thank you for the suggestion, but I end up with almost the same error.

dagster._core.errors.DagsterInvariantViolationError: Core compute for op "dbt_cli_op" returned an output "my_dbt_model" that does not exist. The available outputs are []

Stack Trace:
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_plan.py", line 275, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
,  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 476, in core_dagster_event_sequence_for_step
    for user_event in _step_output_error_checked_user_event_sequence(
,  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 167, in _step_output_error_checked_user_event_sequence
    raise DagsterInvariantViolationError(

[rexledesma]

What dagster-dbt version are you running? This issue should be fixed in dagster-dbt>=0.21.6: #17405.

[emilee-jurkowski]

Ohh, we are using dagster-dbt==0.21.4. Is dagster-dbt>=0.21.6 compatible with dbt 1.5.1? That is the version of dbt that we are currently using.

[rexledesma]

Yes, it is.

[emilee-jurkowski]

Thank you, I will try upgrading.