Hey, I'm trying out a POC of dagster currently and comparing to something like Luigi and Airflow where it is really simple to chain tasks dynamically given the input. Is it possible to do something similar in dagster? I want to do something like:

def run_node(node):
    if node.has_child_nodes():
        for child_node in node.child_nodes():
            yield run_node(child_node)

    yield {"res": my_func(node)}


def my_job():
    graph = load_graph()
    for node in graph:
        run_node(node)

+1 I also would really like to do something like this.

You can use dynamic outputs for this https://docs.dagster.io/concepts/ops-jobs-graphs/dynamic-graphs happy to help answer ay clarifying questions you may have!

Thanks @jamie, are there any runnable examples of using dynamic graphs? None of the snippets in the above docs can actually be run as-is unfortunately. Also, is it possible to combine this with

AssetsDefinition.from_graph

to create assets lineage? It looks like dagster-dbt is doing something similar to what I want -- loading some input/config and building a runnable lineage in dagster from this. I tried reprocuding the dagster-dbt way but I'm not sure how to use

AssetsDefinition

and

AssetKey

correctly. The following minimal example just gives

KeyError

when creating the

AssetsDefinition

for the

keys_by_output_name

from typing import Dict, Tuple
from dagster import (
    AssetKey,
    AssetsDefinition,
    In,
    Nothing,
    Out,
    op,
)

def get_op(op_name, k, ins, outs):
    @op(name=op_name)
    def my_op():
        print(k)

    return my_op


def get_assets() -> AssetsDefinition:

    asset_ins: Dict[AssetKey, Tuple[str, In]] = {}
    asset_outs: Dict[AssetKey, Tuple[str, Out]] = {}

    asset_ins[AssetKey("foo")] = ("some_input", In(Nothing))
    asset_outs[AssetKey("bar")] = ("some_output", Out())

    my_op = get_op(
        op_name="some_output",
        k="foo",
        ins=dict(asset_ins.values()),
        outs=dict(asset_outs.values()),
    )

    return AssetsDefinition(
        keys_by_input_name={
            input_name: asset_key for asset_key, (input_name, _) in asset_ins.items()
        },
        keys_by_output_name={
            output_name: asset_key for asset_key, (output_name, _) in asset_outs.items()
        },
        node_def=my_op,
    )

get_assets()

your code snippets look good! sorry I misunderstood your original request. The Dynamic Outs concept is for when you want to have a graph of ops where you have some top level op that then fans out into an unknown number of downstream ops. But for what you want to do (from what I gather, that’s dynamically construct a set of assets from some external config, but correct me if i’m wrong) is done just like your code snippets. the graphql error you’re seeing is unrelated to your code and is a bug that was introduced in a recent version. If you upgrade to the latest version of dagster, I believe it will go away

Ah thanks for the clarification! I can confirm that the code (my last snippet) works now without any error after upgrading to dagster & dagit 1.1.7. Thanks for the help!