Solid with multiple OutputDefinitions

[william-episource]

I'm working through the tutorials and would like to have multiple type checks added on to the return of my function but keep getting an error, how do I do this?

# I want to do the following below:
@solid(config_schema={"url": str}, output_defs=[OutputDefinition(SimpleDataFrame, name="one"), OutputDefinition(ComplextDataFrame, name="two")])
def download_cereals(context):
    response = requests.get(context.solid_config["url"])
    lines = response.text.split("\n")
    return [row for row in csv.DictReader(lines)]



def less_simple_data_frame_type_check(_, value):
    if not isinstance(value, list):
        return TypeCheck(
            success=False,
            description=f"LessSimpleDataFrame should be a list of dicts, got {type(value)}",
        )

    fields = [field for field in value[0].keys()]

    for i in range(len(value)):
        row = value[i]
        idx = i + 1
        if not isinstance(row, dict):
            return TypeCheck(
                success=False,
                description=(
                    f"LessSimpleDataFrame should be a list of dicts, got {type(row)} for row {idx}"
                ),
            )
        row_fields = [field for field in row.keys()]
        if fields != row_fields:
            return TypeCheck(
                success=False,
                description=(
                    f"Rows in LessSimpleDataFrame should have the same fields, got {row_fields} "
                    f"for row {idx}, expected {fields}"
                ),
            )

    return TypeCheck(
        success=True,
        description="LessSimpleDataFrame summary statistics",
        metadata={
            "n_rows": len(value),
            "n_cols": len(value[0].keys()) if len(value) > 0 else 0,
            "column_names": str(
                list(value[0].keys()) if len(value) > 0 else []
            ),
        },
    )
ComplextDataFrame = DagsterType(
    name="ComplextDataFrame",
    type_check_fn=less_simple_data_frame_type_check,
    description="A metadata checks of data frame, computes statistics",
)


def is_list_of_dicts(_, value):
    return isinstance(value, list) and all(
        isinstance(element, dict) for element in value
    )
SimpleDataFrame = DagsterType(
    name="SimpleDataFrame",
    type_check_fn=is_list_of_dicts,
    description="A naive representation of a data frame, e.g., as returned by csv.DictReader.",
)

Output of error message:

dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline cereal_pipeline, received a tuple of multiple outputs for input "cereals" (at position 0) in solid invocation find_highest_calorie_cereal. Must pass individual output, available from tuple: ('one', 'two')
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/grpc/server.py", line 205, in __init__
    self._repository_symbols_and_code_pointers.load()
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/grpc/server.py", line 91, in load
    self._loadable_target_origin
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/grpc/server.py", line 113, in load_loadable_repository_symbols
    loadable_target_origin.attribute,
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/grpc/utils.py", line 37, in get_loadable_targets
    else loadable_targets_from_python_package(package_name)
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/workspace/autodiscovery.py", line 28, in loadable_targets_from_python_package
    module = load_python_module(package_name, remove_from_path_fn=remove_from_path_fn)
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/code_pointer.py", line 182, in load_python_module
    module = importlib.import_module(module_name)
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1006, in _gcd_import
  File "<frozen importlib._bootstrap>", line 983, in _find_and_load
  File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/Users/william.van/Desktop/epi_repos/dagster_develop/dagster_develop/__init__.py", line 1, in <module>
    from .repository import dagster_develop
  File "/Users/william.van/Desktop/epi_repos/dagster_develop/dagster_develop/repository.py", line 5, in <module>
    from dagster_develop.pipelines.cereal_pipeline import cereal_pipeline
  File "/Users/william.van/Desktop/epi_repos/dagster_develop/dagster_develop/pipelines/cereal_pipeline.py", line 10, in <module>
    @pipeline(mode_defs=[MODE_PROD, MODE_TEST])
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/definitions/decorators/pipeline.py", line 77, in __call__
    ignore_output_from_composition_fn=not self.did_pass_outputs,
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/definitions/composition.py", line 839, in do_composition
    output = fn(**kwargs)
  File "/Users/william.van/Desktop/epi_repos/dagster_develop/dagster_develop/pipelines/cereal_pipeline.py", line 14, in cereal_pipeline
    most_calories=find_highest_calorie_cereal(cereals),
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/definitions/solid.py", line 144, in __call__
    return super(SolidDefinition, self).__call__(*args, **kwargs)
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/definitions/i_solid_definition.py", line 153, in __call__
    )(*args, **kwargs)
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/definitions/composition.py", line 342, in __call__
    "(at position {idx})".format(idx=idx),
  File "/Users/william.van/opt/anaconda3/envs/pytorch/lib/python3.7/site-packages/dagster/core/definitions/composition.py", line 432, in _process_argument_node
    options=output_node._fields,

[alangenfeld]

there is an example for multiple outputs at the solid level here
https://docs.dagster.io/concepts/solids-pipelines/solids#outputs
and pipeline level here
https://docs.dagster.io/concepts/solids-pipelines/pipelines#conditional-branching

[william-episource]

@alangenfeld i would like to ask if i can have a single output (same code)return [row for row in csv.DictReader(lines)] but attach multiple OutputDefinitions? I know I want SimpleDataframe check as well as ComplextDataFrame check to compute the statistics.

Don't want to return/yield multiple items in my actual function.

[alangenfeld]

Ah I see - no this is not possible today. More granular control for what checks get run on inputs/outputs is something we are investigating though.

cc @sryza

[william-episource]

Alrighty thank you for letting me know, can you please close this thread as resolved.