Search code examples
pythondagster

Adding additional parameters to a solid function


I want to add additional parameters when calling a solid, that inherits from another solid as like:

from dagster import pipeline, repository, schedule, solid, InputDefinition, solid

@solid
def hello():
    return 1


@solid(
    input_defs=[
        InputDefinition("name", str),
        InputDefinition("age", int),
    ]
)
def hello2(hello_: int, name: str, age: int):
    print(f"Hello {name}, {age+hello_}")


@pipeline
def pipe_2():
    x = hello()
    y = hello2(x, "Marcus", 20)


@repository
def deploy_docker_repository():
    return [pipe_2, my_schedule]

But when running the pipeline from the dagster API grpc I get the following error:

dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline pipe_2, received invalid
type <class 'str'> for input "name" (at position 1) in solid invocation "hello2".
Must pass the output from 
previous solid invocations or inputs to the composition 
function as inputs when invoking solids during composition.

How to fix?


Solution

  • "Marcus" and 20 need to both be outputs of other solids, for example

    @solid
    def name():
        return "Marcus"
    
    @solid
    def age():
        return 20
    
    @pipeline
    def pipe2():
        hello2(hello(), name(), age())
    

    or else you could pass them in as configuration, i.e. config_schema to hello2 rather than inputs.