Question about parametising reusable solid

[gherka]

Suppose we have a function that looks for a file that could be in different places depending on the given feed type. Although not strictly reusable, it's nice to keep all the logic related to finding this file in one place rather than creating multiple functions for each feed type. From reading this tutorial it would seem that the recommended way is to alias the function, but I'm not sure about how to provide the feed type parameter to it.

@solid(
    name="find_feed_filepath"
)
def find_feed_filepath(context, feed):
    # ...
    # logic to find the correct file based on feed
    
    return True

@pipeline
def demo_pipeline():
    find_feed_filepath.alias("feed_A")()
    find_feed_filepath.alias("feed_B")()

I tried adding config_schema={"feed" : String} to the solid definition, but that means the parameterisation has to happen outside of the @pipeline and feels a bit out of the way for such a simple task. I tried passing feed type directly, but get the error:

In @pipeline demo_pipeline, received invalid type <class 'str'> for input "feed" (at position 0)
in solid invocation "feed_A".  Must pass the output from previous solid invocations or inputs
to the composition function as inputs when invoking solids during composition.

Which explains what is happening really well, but assumes that I want to treat the parameter as input which is not quite the case as this is the first step in the pipeline.

[schrockn]

You might want to try configured, which allows you to curry in config values into a solid definition:

from dagster import pipeline, solid, execute_pipeline


@solid(config_schema={"string": str})
def print_thing(context):
    context.log.info(f"the string: {context.solid_config['string']}")


@pipeline
def pipe():
    print_thing.configured({"string": "foo"}, name="print_foo")()
    print_thing.configured({"string": "bar"}, name="print_bar")()


execute_pipeline(pipe)

[gherka]

This is beautiful, thank you! :)