So in the old API i was able to run a

run_operation

like below. However i cannot seem to figure out how this would look with the new changes to the

DbtCLIResource

?

@asset(
    required_resource_keys={"dbt"},
    op_tags={"kind": "dbt"})
def dbt_stage_external_tables_asset(context):
    return context.resources.dbt.run_operation(macro='stage_external_sources')

Something like:

@asset(
    op_tags={"kind": "dbt"}
)
def dbt_stage_external_tables_asset(context: OpExecutionContext, dbt: DbtCliResource):
    dbt_run_operation_invocation = dbt.cli(["run-operation", "stage_external_sources"]).wait()
    run_results = dbt_run_operations_invocation.get_artifact("run_results.json")

    return run_results

Check out the examples: https://docs.dagster.io/_apidocs/libraries/dagster-dbt#dagster_dbt.DbtCliResource.cli