Configured API#

The configured API offers a way to configure a Dagster entity at definition time.

Relevant APIs#

Overview#

Before we introduce the configured API, to configure a Dagster entity, for example the resource s3_session:

@resource(config_schema={"region": str, "use_unsigned_session": bool})
def s3_session(_init_context):
    """Connect to S3"""

you will need to specify the configurations at runtime via a run config like:

resources:
  key:
    config:
      region: us-east-1
      use_unsigned_session: False

However, in many cases, the option to configure an entity at runtime is more distracting than helpful, and it's preferable to supply the entity's configuration at definition time.

The configured API offers a way to do this. When invoked on a Dagster entity, it returns an interchangeable object with the given configuration "baked in". It is available with the following definitions:

• ResourceDefinition
• ExecutorDefinition
• SolidDefinition
• CompositeSolidDefinition
• LoggerDefinition

Using Configured#

There are different ways to invoke configured on an entity.

As a Method on an Entity#

You can invoke the configured as a method on a given entity.

east_unsigned_s3_session = s3_session.configured(
    {"region": "us-east-1", "use_unsigned_session": False}
)

As a Decorator#

We also provide a configured decorator that makes it easy to create a function-configured version of an object. You can find more information in the @configured API reference.

@configured(s3_session)
def west_unsigned_s3_session(_init_context):
    return {"region": "us-west-1", "use_unsigned_session": False}

As a Standalone API#

If the config to supply to the object is constant, you may alternatively invoke this and call the result with a dict of config values to be curried. You can find more information in the @configured API reference.

west_signed_s3_session = configured(s3_session)(
    {"region": "us-west-1", "use_unsigned_session": False}
)

Examples#

Partially filling the configuration#

In other cases, it's useful to partially fill out the configuration at definition time and leave other configurations for runtime. For these cases, configured can be used as a decorator, accepting a function that translates from runtime config to config that satisfies the entity's config schema. It returns an entity with the "outer" config schema as its schema.

from dagster import configured, resource


@resource(config_schema={"region": str, "use_unsigned_session": bool})
def s3_session(_init_context):
    """Connect to S3"""


@configured(s3_session, config_schema={"region": str})
def unsigned_s3_session(config):
    return {"region": config["region"], "use_unsigned_session": False}

Specifying solid configuration#

You can use the configured API with any definition type in the same way. For example, to configure a solid, you can simply invoke configured on the solid definition:

from dagster import Field, configured, solid


@solid(
    config_schema={"iterations": int, "word": Field(str, is_required=False, default_value="hello")}
)
def example_solid(context):
    for _ in range(context.solid_config["iterations"]):
        context.log.info(context.solid_config["word"])


# This example is fully configured. With this syntax, a name must be explicitly provided.
configured_example = configured(example_solid, name="configured_example")(
    {"iterations": 6, "word": "wheaties"}
)

# This example is partially configured: `iterations` is passed through
# The decorator yields a solid named 'another_configured_example' (from the decorated function)
# with `int` as the `config_schema`.
@configured(example_solid, int)
def another_configured_example(config):
    return {"iterations": config, "word": "wheaties"}

Patterns#

Reusing a solid definition with configured#

When using the decorator syntax (@configured), the resulting solid definition will inherit the name of the function being decorated (like another_configured_example in the above example). When configuring a solid completely with a config dictionary rather than with a function (as with configured_example), you must add the positional argument name in the call to configured. When naming solids, remember that solid definitions must have unique names within a repository or pipeline.

@solid(
    config_schema={
        "is_sample": Field(bool, is_required=False, default_value=False),
    },
    input_defs=[InputDefinition("xs", List[Int])],
)
def get_dataset(context, xs):
    if context.solid_config["is_sample"]:
        return xs[:5]
    else:
        return xs


# If we want to use the same solid configured in multiple ways in the same pipeline,
# we have to specify unique names when configuring them:
sample_dataset = configured(get_dataset, name="sample_dataset")({"is_sample": True})
full_dataset = configured(get_dataset, name="full_dataset")({"is_sample": False})


@pipeline
def dataset_pipeline():
    sample_dataset()
    full_dataset()