mapper.py

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

@solid(input_defs=[InputDefinition("number", int)])
def process(context, number):
    context.log.info("Number: {}".format(number))
    
@solid
def root(context):
    numbers = [1, 2, 3, 4, 5]
    for number in numbers:
        execute_pipeline(process_number, instance=DagsterInstance.get(), environment_dict={
            'solids': {
                'my_solid': {
                    'inputs': {
                        'number': number
                    }
                }
            }
        })


@pipeline
def root_pipeline():
    root()

@pipeline
def process_number():
    process()

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