Hi all. Is it possible to have an input parameter of type

Dict

on a job level? I wonder if I'm doing something wrong, since when when I'm trying to run this job:

from typing import Any, Dict

from dagster import job, op
from dagster_aws.s3 import s3_pickle_io_manager, s3_resource


@op
def test_op(context, parameters_dict: Dict[str, Any]):
    <http://context.log.info|context.log.info>(parameters_dict)


@job(resource_defs={
    "io_manager": s3_pickle_io_manager,
    "s3": s3_resource,
})
def test_job(parameters_dict: Dict[str, Any]):
    test_op(parameters_dict)

With the following parameters:

inputs:
  parameters_dict:
    param_1: value1
    param_2: 2
resources:
  io_manager:
    config:
      s3_bucket: my-bucket

I'm getting the following error:

dagster.core.errors.DagsterTypeLoadingError: Error occurred while loading input "parameters_dict" of step "test_op":
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_plan.py", line 224, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_step.py", line 316, in core_dagster_event_sequence_for_step
    step_input.source.load_input_object(step_context, input_def)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/inputs.py", line 627, in load_input_object
    return dagster_type.loader.construct_from_config_value(step_context, config_data)
  File "/usr/lib/python3.8/contextlib.py", line 131, in __exit__
    self.gen.throw(type, value, traceback)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/errors.py", line 191, in user_code_error_boundary
    raise error_cls(
The above exception was caused by the following exception:
AttributeError: 'str' object has no attribute 'items'
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/errors.py", line 184, in user_code_error_boundary
    yield
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/inputs.py", line 627, in load_input_object
    return dagster_type.loader.construct_from_config_value(step_context, config_data)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/python_dict.py", line 38, in construct_from_config_value
    runtime_value[key] = self._value_dagster_type.loader.construct_from_config_value(
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/config_schema.py", line 156, in construct_from_config_value
    return self._func(context, config_value)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/builtin_config_schemas.py", line 52, in _any_input_schema
    return load_type_input_schema_dict(config_value)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/builtin_config_schemas.py", line 35, in load_type_input_schema_dict
    file_type, file_options = list(value.items())[0]

And with scalar inputs the same code works just fine

Hi Grigoriy. We don't currently support inputs to a job, though an op within a job can accept an input via config. If you redefined your job to accept the dict input via config, this would work:

@op(ins={"parameters_dict": In(Dict[str, int])})
def test_op(context, parameters_dict: Dict[str, Any]):
    <http://context.log.info|context.log.info>(parameters_dict)


@job
def test_job():
    test_op()


test_job.execute_in_process(
    run_config={"ops": {"test_op": {"inputs": {"parameters_dict": {"foo": 5}}}}}
)

Hi @claire . But for some reason my example with job level inputs works with scalar type inputs. Also job decorator has an input_defs parameter for which the Python docstring states that job level inputs are supported

Ah--my mistake. Then job level inputs are supported

But only scalar types? Or is it a bug?

Yep, this is a bug. I will file an issue for this:

Created issue at: https://github.com/dagster-io/dagster/issues/9808

Thank you!