recursive_pipeline.py

from dagster import pipeline, solid, RepositoryDefinition, InputDefinition, execute_pipeline_iterator, DagsterInstance


@solid(input_defs=[InputDefinition("number", int)])
def my_solid(context, number):
    context.log.info("Number: {}".format(number))

    # Stop condition to prevent infinite recursion
    if number == 5:
        return

    for event in execute_pipeline_iterator(my_pipeline, instance=DagsterInstance.get(), environment_dict={
        'solids': {
            'my_solid': {
                'inputs': {
                    'number': number + 1
                }
            }
        }
    }):
        context.log.info(str(event)) # or simply call pass


@pipeline
def my_pipeline():
    my_solid()


def define_repository():
    return RepositoryDefinition("my_repository", pipeline_defs=[my_pipeline])

Alternatively, using execute_pipeline:

from dagster import pipeline, solid, RepositoryDefinition, InputDefinition, execute_pipeline, DagsterInstance


@solid(input_defs=[InputDefinition("number", int)])
def my_solid(context, number):
    context.log.info("Number: {}".format(number))

    # Stop condition to prevent infinite recursion
    if number == 5:
        return

    execute_pipeline(my_pipeline, instance=DagsterInstance.get(), environment_dict={
        'solids': {
            'my_solid': {
                'inputs': {
                    'number': number + 1
                }
            }
        }
    })


@pipeline
def my_pipeline():
    my_solid()


def define_repository():
    return RepositoryDefinition("my_repository", pipeline_defs=[my_pipeline])