Proper use of @graph_multi_asset?

[stalerik95]

Hi there!

I'm new to Dagster, am developing a solution combining it with dbt. I want a multi-asset that materializes all assets for a dbt source (a collection of tables, presumed to exist by the dbt project). I'd like to be able to materialize a subset of the assets, wherefore I need access to the execution context and context.selected_asset_keys. I've understood this cannot be done neither within @graph nor @graph_multi_asset.

As a beginner, I know I shouldn't be dabbling with graphs, and just use @multi_asset instead, however I'm tempted to use @graph_multi_asset, as that allows me to see the sub-processes which the asset creation consists of.

Attempt 1

from dagster import graph_multi_asset, graph, AssetOut, MaterializeResult, op, OpExecutionContext
from dagster_dbt import get_asset_keys_by_output_name_for_source
from ..assets import dbt_assets

@op
def my_op(context: OpExecutionContext):
    results = {}
    for asset_key in context.selected_asset_keys:
        load_data(asset_key.to_string())
        results[asset_key.to_string()] = MaterializeResult(asset_key=asset_key)

@graph
def my_landing_graph() -> dict:
    return my_op()

@graph_multi_asset(
        outs={
            name: AssetOut(key=asset_key)
            for name, asset_key in get_asset_keys_by_output_name_for_source(
                [dbt_assets], "kp_landing"
            ).items()
        },
        can_subset=True
    )
def my_landing_graph_multi_asset():
    return my_landing_graph()

This yeilds the following error:

dagster._core.errors.DagsterInvalidDefinitionError: Returned a single output (my_landing_graph.result) in @graph 'my_landing_graph_multi_asset' but 20 outputs are defined. Return a dict to map defined outputs.

The results from my_landing_graph() are already in a dictionary format, so the error puzzles me. As a mitigation I tried repacking the dictionary as follows:

Attempt 2

from dagster import graph_multi_asset, graph, AssetOut, MaterializeResult, op, OpExecutionContext
from dagster_dbt import get_asset_keys_by_output_name_for_source
from ..assets import dbt_assets

@op
def my_op(context: OpExecutionContext):
    results = {}
    for asset_key in context.selected_asset_keys:
        load_data(asset_key.to_string())
        results[asset_key.to_string()] = MaterializeResult(asset_key=asset_key)

@graph
def my_landing_graph() -> dict:
    return my_op()

@graph_multi_asset(
        outs={
            name: AssetOut(key=asset_key)
            for name, asset_key in get_asset_keys_by_output_name_for_source(
                [dbt_assets], "kp_landing"
            ).items()
        },
        can_subset=True
    )
def my_landing_graph_multi_asset():
    results = my_landing_graph()
    return {name: results[name] for name in my_landing_graph()}

This yeilds the following error (I have no idea why it adds a "_2" to the name):

dagster._core.errors.DagsterInvariantViolationError: Attempted to iterate over an InvokedNodeOutputHandle. This object represents the output "result" from the op/graph "my_landing_graph_2". Consider defining multiple Outs if you seek to pass different parts of this output to different op/graph.

I understand that both the @graph and @graph_multi_asset functions are run during the build phase and not during execution. Therefore it somehow makes sense that my_landing_graph returns a InvokedNodeOutputHandle instead of the actual dictionary, however my_landing_graph_multi_asset expects a dictionary, so I'm unsure what to do instead. I've tried reading the documentation, and have asked a few LLM's, but haven't found a solution. Any help is appreciated!

[matthias-Q]

I have the same issue. Would be great to have better examples for graph_multi_asset