import makefun
import dagster
import inspect

def solid_from_func(func):
    # capture the signature of func
    sig = inspect.signature(func)
    params = list(sig.parameters.values())
    params.insert(0, inspect.Parameter('context', kind=inspect.Parameter.POSITIONAL_OR_KEYWORD))
    new_sig = sig.replace(parameters=params)

    # modify the signature to add 'context'
    params = list(sig.parameters.values())

    type_map = { 
        bool: dagster.Bool,
        float: dagster.Float,
        int: <http://dagster.Int|dagster.Int>,
        str: dagster.String,
        list: dagster.Array,
        dict: dagster.Dict
    }

    config_from_params = {}

    for k, v in sig.parameters.items():
        try:
            if v.default is not inspect.Parameter.empty:
                config_from_params.update({k: dagster.Field(type_map[type(v.default)], default_value=v.default, is_optional=True)})
            else:
                config_from_params.update({k: dagster.Field(type_map[type(v.default)], is_optional=False)})
        except KeyError:
            config_from_params.update({k: dagster.Field(dagster.Any, is_optional=False)})
    
    # create our solid
    @dagster.solid(config=config_from_params)
    @makefun.wraps(func, new_sig=new_sig)
    def new_solid(context, *args, **kwargs):
        # call the original function
        return func(context, *args, **kwargs)

    return new_solid

decorator_name = '@solid', fn_name = 'read_excel', compute_fn = <function read_excel at 0x000001BED642FD38>
input_defs = [<dagster.core.definitions.input.InputDefinition object at 0x000001BED643DA08>, <dagster.core.definitions.input.InputD...tion object at 0x000001BED643DAC8>, <dagster.core.definitions.input.InputDefinition object at 0x000001BED644C848>, ...]
expected_positionals = ['context'], exclude_nothing = True

    def validate_solid_fn(
        decorator_name, fn_name, compute_fn, input_defs, expected_positionals=None, exclude_nothing=True
    ):
        check.str_param(decorator_name, 'decorator_name')
        check.str_param(fn_name, 'fn_name')
        check.callable_param(compute_fn, 'compute_fn')
        check.list_param(input_defs, 'input_defs', of_type=InputDefinition)
        expected_positionals = check.opt_list_param(
            expected_positionals, 'expected_positionals', of_type=str
        )
        if exclude_nothing:
            names = set(inp.name for inp in input_defs if not inp.runtime_type.is_nothing)
            nothing_names = set(inp.name for inp in input_defs if inp.runtime_type.is_nothing)
        else:
            names = set(inp.name for inp in input_defs)
            nothing_names = set()

        # Currently being super strict about naming. Might be a good idea to relax. Starting strict.
        fn_positionals, input_args = split_function_parameters(compute_fn, expected_positionals)

        # Validate Positional Parameters
        missing_positional = validate_decorated_fn_positionals(fn_positionals, expected_positionals)
        if missing_positional:
            raise DagsterInvalidDefinitionError(
                "{decorator_name} '{solid_name}' decorated function does not have required positional "
                "parameter '{missing_param}'. Solid functions should only have keyword arguments "
                "that match input names and a first positional parameter named 'context'.".format(
>                   decorator_name=decorator_name, solid_name=fn_name, missing_param=missing_positional
                )
            )
E           dagster.core.errors.DagsterInvalidDefinitionError: @solid 'read_excel' decorated function does not have required positional parameter 'context'. Solid functions should only have keyword arguments that match input names and a first positional parameter named 'context'.

venv\lib\site-packages\dagster\core\definitions\decorators.py:494: DagsterInvalidDefinitionError

makefun.wraps

# create our solid
    @dagster.solid(
        name=func.__name__,
        config=config_from_params,
    )
    def new_solid(context):
        # call the original function with the config values
        return func(**context.solid_config)

input_values

environment_dict

execute_solid