How to specify upstream asset dependencies when using AssetsDefinition.from_graph() ?

[dagsir[bot]]

Hey, is it possible to speciy upstream asset depedencies when using AssetsDefinition.from_graph() ?

---

The question was originally asked in Dagster Slack.

[benpankow]

Yes - inputs to the graph are assumed to map to upstream asset dependencies. For example, given the following graph:

@op(ins={"usernames": In(List[str])})
def process_users() -> List[str]:
    ...


@op(ins={"transatctions": In(Dict[str, int])})
def process_transactions() -> Dict[str, str]:
    ...


@op(
    ins={"processed_users": In(List[str]), "processed_transactions": In(Dict[str, str])},
    out={"results": Out(int)}
)
def process_results() -> int:
    ...


@graph
def my_graph():
    return process_results(process_users(), process_transactions())

You can attach asset keys to the inputs as follows:

assets = AssetsDefinition.from_graph(
    my_graph,
    keys_by_input_name={
        "usernames": AssetKey("usernames"),
        "transactions": AssetKey("transactions"),
    },
    keys_by_output_name={"results": AssetKey("processed_results")},
)

[axellpadilla]

I tried this but get

dagster._check.functions.CheckError: Invariant failed. Description: The set of input names keys specified in the keys_by_input_name argument must equal the set of asset keys inputted by 'graph'. keys_by_input_name keys: { 'my_keys'' }
expected keys: []

[axellpadilla]

I tried this but get

dagster._check.functions.CheckError: Invariant failed. Description: The set of input names keys specified in the keys_by_input_name argument must equal the set of asset keys inputted by 'graph'. keys_by_input_name keys: { 'my_keys'' }
expected keys: []

Fixed adding all Ins also to the graph, and from the graph using **kwargs and adding each asset to each op.