Allow passing arguments to map on DynamicOutput

[ivanov-slk]

Currently one is able to invoke a solid on each element of an Iterable like this:

@solid
def return_some_list(context):
    return [1, 2, 3, 4, 5]


@solid(output_defs=[DynamicOutputDefinition(int)])
def generate_subtasks(context, nums: List[int]):
    context.log.info(str(nums))
    for num in nums:
        yield DynamicOutput(num, mapping_key=f'subtask_{num}')


@solid(input_defs=[
    InputDefinition('some_num', int),
])
def print_num(context, some_num: int):
    context.log.info(str(some_num))
    return some_num

@pipeline
def some_pipeline():
    output_list = return_some_list()
    out = generate_subtasks(output_list).map(print_num)

However, this allows the callable solid (print_num in this case) to accept only one argument. If I'm not wrong, this is specified in dagster.core.definitions.composition.InvokedSolidDynamicOutputWrapper.map, where the map method accepts only a callable. This is inconvenient if the mappable function requires additional arguments set at runtime (i.e. outside of the config_schema).

A possible solution would look in the lines of:

...
@solid(input_defs=[
    InputDefinition('some_num', int),
    InputDefinition('arg1', str),
    InputDefinition('arg2', str)
])
def print_num(context, some_num: int, arg1, arg2):
    context.log.info(str(some_num) + arg1 + arg2)
    raise Exception
    return some_num


@pipeline
def some_pipeline():
    output_list = return_some_list()
    out = generate_subtasks(output_list).map(print_num, 'print', 'this')

[alangenfeld]

Thanks for the idea, we'll take this in to consideration.

Just to note, in the mean time some ways to deal with this are to use lambda or functools.partial

@pipeline
def lambda():
    output_list = return_some_list()
    print, this = return_strings()
    out = generate_subtasks(output_list).map(lambda num: print_num(num, print, this))

@pipeline
def partial():
    output_list = return_some_list()
    print, this = return_strings()
    partial_print_num = partial(print_num, arg1=print, arg2=this)
    out = generate_subtasks(output_list).map(partial_print_num)