Config

Pythonic config system

The following classes are used as part of the new Pythonic config system. They are used in conjunction with builtin types.

class dagster.Config

Base class for Dagster configuration models, used to specify config schema for ops and assets. Subclasses pydantic.BaseModel.

Example definition:

from pydantic import Field

class MyAssetConfig(Config):
    my_str: str = "my_default_string"
    my_int_list: List[int]
    my_bool_with_metadata: bool = Field(default=False, description="A bool field")

Example usage:

@asset
def asset_with_config(config: MyAssetConfig):
    assert config.my_str == "my_default_string"
    assert config.my_int_list == [1, 2, 3]
    assert config.my_bool_with_metadata == False

asset_with_config(MyAssetConfig(my_int_list=[1, 2, 3], my_bool_with_metadata=True))

class dagster.PermissiveConfig

Subclass of ConfigConfig that allows arbitrary extra fields. This is useful for config classes which may have open-ended inputs.

Example definition:

class MyPermissiveOpConfig(PermissiveConfig):
    my_explicit_parameter: bool
    my_other_explicit_parameter: str

Example usage:

@op
def op_with_config(config: MyPermissiveOpConfig):
    assert config.my_explicit_parameter == True
    assert config.my_other_explicit_parameter == "foo"
    assert config.dict().get("my_implicit_parameter") == "bar"

op_with_config(
    MyPermissiveOpConfig(
        my_explicit_parameter=True,
        my_other_explicit_parameter="foo",
        my_implicit_parameter="bar"
    )
)

class dagster.RunConfig

Container for all the configuration that can be passed to a run. Accepts Pythonic definitions for op and asset config and resources and converts them under the hood to the appropriate config dictionaries.

Example usage:

class MyAssetConfig(Config):
    a_str: str

@asset
def my_asset(config: MyAssetConfig):
    assert config.a_str == "foo"

materialize(
    [my_asset],
    run_config=RunConfig(
        ops=\{"my_asset": MyAssetConfig(a_str="foo")}
    )
)

to_config_dict

Converts the RunConfig to a dictionary representation.

Returns: The dictionary representation of the RunConfig.Return type: Dict[str, Any]

Legacy Dagster config types

The following types are used as part of the legacy Dagster config system. They are used in conjunction with builtin types.

class dagster.ConfigSchema

Placeholder type for config schemas.

Any time that it appears in documentation, it means that any of the following types are acceptable:

1. A Python scalar type that resolves to a Dagster config type (python:int, python:float, python:bool, or python:str). For example:

   • @op(config_schema=int)
   • @op(config_schema=str)

2. A built-in python collection (python:list, or python:dict). python:list is exactly equivalent to ArrayArray [ Any ] and python:dict is equivalent to PermissivePermissive. For example:

   • @op(config_schema=list)
   • @op(config_schema=dict)

3. A Dagster config type:

   • Any
   • ArrayArray
   • Bool
   • EnumEnum
   • Float
   • Int
   • IntSourceIntSource
   • NoneableNoneable
   • PermissivePermissive
   • MapMap
   • ScalarUnionScalarUnion
   • SelectorSelector
   • ShapeShape
   • String
   • StringSourceStringSource

4. A bare python dictionary, which will be automatically wrapped in ShapeShape. Values of the dictionary are resolved recursively according to the same rules. For example:

   • \{'some_config': str} is equivalent to Shape(\{'some_config: str}).
   • \{'some_config1': \{'some_config2': str}} is equivalent to

5. A bare python list of length one, whose single element will be wrapped in a ArrayArray is resolved recursively according to the same rules. For example:

   • [str] is equivalent to Array[str].
   • [[str]] is equivalent to Array[Array[str]].
   • [\{'some_config': str}] is equivalent to Array(Shape(\{'some_config: str})).

6. An instance of FieldField.

class dagster.Field

Defines the schema for a configuration field.

Fields are used in config schema instead of bare types when one wants to add a description, a default value, or to mark it as not required.

Config fields are parsed according to their schemas in order to yield values available at job execution time through the config system. Config fields can be set on ops, on loaders for custom, and on other pluggable components of the system, such as resources, loggers, and executors.

Parameters:

• config (Any) –

  The schema for the config. This value can be any of:

  1. A Python primitive type that resolves to a Dagster config type (python:int, python:float, python:bool, python:str, or python:list).

  2. A Dagster config type:

     • Any
     • ArrayArray
     • Bool
     • EnumEnum
     • Float
     • Int
     • IntSourceIntSource
     • NoneableNoneable
     • PermissivePermissive
     • ScalarUnionScalarUnion
     • SelectorSelector
     • ShapeShape
     • String
     • StringSourceStringSource

  3. A bare python dictionary, which will be automatically wrapped in ShapeShape. Values of the dictionary are resolved recursively according to the same rules.

  4. A bare python list of length one which itself is config type.

• default_value (Any) –

  A default value for this field, conformant to the schema set by the dagster_type argument. If a default value is provided, is_required should be False.

• is_required (bool) – Whether the presence of this field is required. Defaults to true. If is_required

• description (str) – A human-readable description of this config field.

Examples:

@op(
    config_schema=\{
        'word': Field(str, description='I am a word.'),
        'repeats': Field(Int, default_value=1, is_required=False),
    }
)
def repeat_word(context):
    return context.op_config['word'] * context.op_config['repeats']

property default_provided

Was a default value provided.

Returns: Yes or noReturn type: bool

property default_value

The default value for the field.

Raises an exception if no default value was provided.

property description

A human-readable description of this config field, if provided.

property is_required

Whether a value for this field must be provided at runtime.

Cannot be True if a default value is provided.

class dagster.Selector

Define a config field requiring the user to select one option.

Selectors are used when you want to be able to present several different options in config but allow only one to be selected. For example, a single input might be read in from either a csv file or a parquet file, but not both at once.

Note that in some other type systems this might be called an ‘input union’.

Functionally, a selector is like a Dict, except that only one key from the dict can be specified in valid config.

Parameters: fields (Dict[str, FieldField]) – The fields from which the user must select. Examples:

@op(
    config_schema=Field(
        Selector(
            \{
                'haw': \{'whom': Field(String, default_value='honua', is_required=False)},
                'cn': \{'whom': Field(String, default_value='世界', is_required=False)},
                'en': \{'whom': Field(String, default_value='world', is_required=False)},
            }
        ),
        is_required=False,
        default_value=\{'en': \{'whom': 'world'}},
    )
)
def hello_world_with_default(context):
    if 'haw' in context.op_config:
        return 'Aloha \{whom}!'.format(whom=context.op_config['haw']['whom'])
    if 'cn' in context.op_config:
        return '你好, \{whom}!'.format(whom=context.op_config['cn']['whom'])
    if 'en' in context.op_config:
        return 'Hello, \{whom}!'.format(whom=context.op_config['en']['whom'])

class dagster.Permissive

Defines a config dict with a partially specified schema.

A permissive dict allows partial specification of the config schema. Any fields with a specified schema will be type checked. Other fields will be allowed, but will be ignored by the type checker.

Parameters: fields (Dict[str, FieldField]) – The partial specification of the config dict. Examples:

@op(config_schema=Field(Permissive(\{'required': Field(String)})))
def map_config_op(context) -> List:
    return sorted(list(context.op_config.items()))

class dagster.Shape

Schema for configuration data with string keys and typed values via FieldField.

Unlike PermissivePermissive, unspecified fields are not allowed and will throw a DagsterInvalidConfigErrorDagsterInvalidConfigError.

Parameters:

• fields (Dict[str, FieldField]) – The specification of the config dict.
• field_aliases (Dict[str, str]) – Maps a string key to an alias that can be used instead of the original key. For example,

class dagster.Map

Defines a config dict with arbitrary scalar keys and typed values.

A map can contrain arbitrary keys of the specified scalar type, each of which has type checked values. Unlike ShapeShape and PermissivePermissive, scalar keys other than strings can be used, and unlike PermissivePermissive, all values are type checked.

Parameters:

• key_type (type) – The type of keys this map can contain. Must be a scalar type.
• inner_type (type) – The type of the values that this map type can contain.
• key_label_name (string) – Optional name which describes the role of keys in the map.

Examples:

@op(config_schema=Field(Map(\{str: int})))
def partially_specified_config(context) -> List:
    return sorted(list(context.op_config.items()))

property key_label_name

Name which describes the role of keys in the map, if provided.

class dagster.Array

Defines an array (list) configuration type that contains values of type inner_type.

Parameters: inner_type (type) – The type of the values that this configuration type can contain.

property description

A human-readable description of this Array type.

class dagster.Noneable

Defines a configuration type that is the union of NoneType and the type inner_type.

Parameters: inner_type (type) – The type of the values that this configuration type can contain. Examples:

config_schema=\{"name": Noneable(str)}

config=\{"name": "Hello"}  # Ok
config=\{"name": None}     # Ok
config=\{}                 # Error

class dagster.Enum

Defines a enum configuration type that allows one of a defined set of possible values.

Parameters:

• name (str) – The name of the enum configuration type.
• enum_values (List[EnumValueEnumValue]) – The set of possible values for the enum configuration type.

Examples:

@op(
    config_schema=Field(
        Enum(
            'CowboyType',
            [
                EnumValue('good'),
                EnumValue('bad'),
                EnumValue('ugly'),
            ]
        )
    )
)
def resolve_standoff(context):
    # ...

class dagster.EnumValue

Define an entry in a EnumEnum.

Parameters:

• config_value (str) – The string representation of the config to accept when passed.
• python_value (Optional[Any]) – The python value to convert the enum entry in to. Defaults to the config_value.
• description (Optional[str]) – A human-readable description of the enum entry.

class dagster.ScalarUnion

Defines a configuration type that accepts a scalar value OR a non-scalar value like a List, Dict, or SelectorSelector.

This allows runtime scalars to be configured without a dictionary with the key value and instead just use the scalar value directly. However this still leaves the option to load scalars from a json or pickle file.

Parameters:

• scalar_type (type) – The scalar type of values that this configuration type can hold. For example,
• non_scalar_schema (ConfigSchemaConfigSchema) – The schema of a non-scalar Dagster configuration type. For example, List,
• key (Optional[str]) – The configuation type’s unique key. If not set, then the key will be set to

Examples:

graph:
  transform_word:
    inputs:
      word:
        value: foobar

becomes, optionally,

graph:
  transform_word:
    inputs:
      word: foobar

dagster.StringSource

Use this type when you want to read a string config value from an environment variable. The value passed to a config field of this type may either be a string literal, or a selector describing how to look up the value from the executing process’s environment variables.

Examples:

from dagster import job, op, StringSource

@op(config_schema=StringSource)
def secret_op(context) -> str:
    return context.op_config

@job
def secret_job():
    secret_op()

secret_job.execute_in_process(
    run_config=\{
        'ops': \{'secret_op': \{'config': 'test_value'}}
    }
)

secret_job.execute_in_process(
    run_config=\{
        'ops': \{'secret_op': \{'config': \{'env': 'VERY_SECRET_ENV_VARIABLE'}}}
    }
)

dagster.IntSource

Use this type when you want to read an integer config value from an environment variable. The value passed to a config field of this type may either be a integer literal, or a selector describing how to look up the value from the executing process’s environment variables.

Examples:

from dagster import job, op, IntSource

@op(config_schema=IntSource)
def secret_int_op(context) -> int:
    return context.op_config

@job
def secret_job():
    secret_int_op()

secret_job.execute_in_process(
    run_config=\{
        'ops': \{'secret_int_op': \{'config': 1234}}
    }
)

secret_job.execute_in_process(
    run_config=\{
        'ops': \{'secret_int_op': \{'config': \{'env': 'VERY_SECRET_ENV_VARIABLE_INT'}}}
    }
)

dagster.BoolSource

Use this type when you want to read an boolean config value from an environment variable. The value passed to a config field of this type may either be a boolean literal, or a selector describing how to look up the value from the executing process’s environment variables. Set the value of the corresponding environment variable to "" to indicate False.

Examples:

from dagster import job, op, BoolSource

@op(config_schema=BoolSource)
def secret_bool_op(context) -> bool:
    return context.op_config

@job
def secret_job():
    secret_bool_op()

secret_job.execute_in_process(
    run_config=\{
        'ops': \{'secret_bool_op': \{'config': False}}
    }
)

secret_job.execute_in_process(
    run_config=\{
        'ops': \{'secret_bool_op': \{'config': \{'env': 'VERY_SECRET_ENV_VARIABLE_BOOL'}}}
    }
)

Config Utilities

class dagster.ConfigMapping

Defines a config mapping for a graph (or job).

By specifying a config mapping function, you can override the configuration for the child ops and graphs contained within a graph.

Config mappings require the configuration schema to be specified as config_schema, which will be exposed as the configuration schema for the graph, as well as a configuration mapping function, config_fn, which maps the config provided to the graph to the config that will be provided to the child nodes.

Parameters:

• config_fn (Callable[[dict], dict]) – The function that will be called
• config_schema (ConfigSchemaConfigSchema) – The schema of the graph config.
• receive_processed_config_values (Optional[bool]) – If true, config values provided to the config_fn

@dagster.configured

A decorator that makes it easy to create a function-configured version of an object.

The following definition types can be configured using this function:

• GraphDefinitionGraphDefinition
• ExecutorDefinitionExecutorDefinition
• LoggerDefinitionLoggerDefinition
• ResourceDefinitionResourceDefinition
• OpDefinitionOpDefinition

Using configured may result in config values being displayed in the Dagster UI, so it is not recommended to use this API with sensitive values, such as secrets.

If the config that will be supplied to the object is constant, you may alternatively invoke this and call the result with a dict of config values to be curried. Examples of both strategies below.

Parameters:

• configurable (ConfigurableDefinition) – An object that can be configured.
• config_schema (ConfigSchemaConfigSchema) – The config schema that the inputs to the decorated function
• **kwargs – Arbitrary keyword arguments that will be passed to the initializer of the returned

Returns: (Callable[[Union[Any, Callable[[Any], Any]]], ConfigurableDefinition]) Examples:

class GreetingConfig(Config):
    message: str

@op
def greeting_op(config: GreetingConfig):
    print(config.message)

class HelloConfig(Config):
    name: str

@configured(greeting_op)
def hello_op(config: HelloConfig):
    return GreetingConfig(message=f"Hello, \{config.name}!")

dev_s3 = configured(S3Resource, name="dev_s3")(\{'bucket': 'dev'})

@configured(S3Resource)
def dev_s3(_):
    return \{'bucket': 'dev'}

@configured(S3Resource, \{'bucket_prefix', str})
def dev_s3(config):
    return \{'bucket': config['bucket_prefix'] + 'dev'}