How do I define a method using dbt.cli that executes run-operation stage_external_sources?

[Iuke]

we have a definition:

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", f"--target={profile_target}"], context=context).stream()

and have tried implementing operations in a few ways below:

OpA

@op()
def dbt_run_operation(context: OpExecutionContext):
    dbt.cli(["run-operation", f"stage_external_sources"], context=context)
    return True

OpB

@op(required_resource_keys={"dbt"})
def dbt_run_operation(context, operation_name: str):
    dbt_operation_cmd = ["run-operation", operation_name]
    context.resources.dbt.cli(dbt_operation_cmd).stream()

@job(resource_defs={"dbt": dbt})
def dbt_run_operation_job():
    dbt_run_operation()

When executing the jobs on our instance we can see logs like this:

17:02:46.834 dbt_run_operation Started execution of step "dbt_run_operation".
17:02:46.849 dbt_run_operation INFO Copying `/home/data-dbt/dbt/target/partial_parse.msgpack` to `/home/data-dbt/dbt/target/dbt_run_operation-d2079a4-9ad3956/partial_parse.msgpack` to take advantage of partial parsing.
17:02:46.867 dbt_run_operation INFO Running dbt command: `dbt run-operation stage_external_sources --profile foo --target foo`.

But the actual refresh of the external sources doesn't occur on the database.

If you can provide a block example of how run-operation, or really any other non @dbt_asset decorated function that calls a dbt function should be defined, it would be helpful - thanks!

[NoamMosko]

joining the question

[gibsondan]

Hi @Iuke - if you run the same command yourself locally via the dbt cli command directly, leaving dagster out of the picture, does it do the operation that you expect?

[Iuke]

yes - running with native dbt creates the external tables as we would expect - so far in dagster-dbt we've been able to materialize our dbt assets within dagster using the @dbt_assets() decorator, but haven't been able to see success with the other cli cmds

[gibsondan]

Note that stream returns a Python generator so that yield from is important to make sure it executes - could that be the problem?

So instead of:

@op(required_resource_keys={"dbt"})
def dbt_run_operation(context, operation_name: str):
    dbt_operation_cmd = ["run-operation", operation_name]
    context.resources.dbt.cli(dbt_operation_cmd).stream()

Try:

@op(required_resource_keys={"dbt"})
def dbt_run_operation(context, operation_name: str):
    dbt_operation_cmd = ["run-operation", operation_name]
    yield from context.resources.dbt.cli(dbt_operation_cmd).stream()

[Iuke]

My teammate @emilee-jurkowski found this related issue and implemented your method and added the out arg.

@op(required_resource_keys={"dbt"}, out=Out(Nothing))
def dbt_run_operation(context: OpExecutionContext, operation_name: str):
    dbt_operation_cmd = ["run-operation", operation_name]
    yield from context.resources.dbt.cli(dbt_operation_cmd).stream()

I would encourage other people who stumble across this discussion to explore different out returns if you need some logging but I think this is good enough for what we need.