IO Managers¶

Input and Output Contexts¶

class dagster.InputContext(name=None, pipeline_name=None, solid_def=None, config=None, metadata=None, upstream_output=None, dagster_type=None, log_manager=None, resource_config=None, resources=None, step_context=None, op_def=None)[source]¶

The context object available to the load_input method of RootInputManager.

name¶

The name of the input that we’re loading.

Type

Optional[str]

pipeline_name¶

The name of the pipeline.

Type

Optional[str]

solid_def¶

The definition of the solid that’s loading the input.

Type

Optional[SolidDefinition]

config¶

The config attached to the input that we’re loading.

Type

Optional[Any]

metadata¶

A dict of metadata that is assigned to the InputDefinition that we’re loading for.

Type

Optional[Dict[str, Any]]

upstream_output¶

Info about the output that produced the object we’re loading.

Type

Optional[OutputContext]

dagster_type¶

The type of this input.

Type

Optional[DagsterType]

log¶

The log manager to use for this input.

Type

Optional[DagsterLogManager]

resource_config¶

The config associated with the resource that initializes the RootInputManager.

Type

Optional[Dict[str, Any]]

resources¶

The resources required by the resource that initializes the input manager. If using the @root_input_manager() decorator, these resources correspond to those requested with the required_resource_keys parameter.

Type

Optional[Resources]

op_def¶

The definition of the op that’s loading the input.

Type

Optional[OpDefinition]

property asset_partition_key¶

The partition key for input asset.

Raises an error if the input asset has no partitioning, or if the run covers a partition range for the input asset.

property asset_partition_key_range¶

The partition key range for input asset.

Raises an error if the input asset has no partitioning.

property asset_partitions_time_window¶

The time window for the partitions of the input asset.

Raises an error if either of the following are true: - The input asset has no partitioning. - The input asset is not partitioned with a TimeWindowPartitionsDefinition.

property has_input_name¶

If we’re the InputContext is being used to load the result of a run from outside the run, then it won’t have an input name.

property has_partition_key¶

Whether the current run is a partitioned run

property partition_key¶

The partition key for the current run.

Raises an error if the current run is not a partitioned run.

class dagster.OutputContext(step_key=None, name=None, pipeline_name=None, run_id=None, metadata=None, mapping_key=None, config=None, solid_def=None, dagster_type=None, log_manager=None, version=None, resource_config=None, resources=None, step_context=None, op_def=None)[source]¶

The context object that is available to the handle_output method of an IOManager.

step_key¶

The step_key for the compute step that produced the output.

Type

Optional[str]

name¶

The name of the output that produced the output.

Type

Optional[str]

pipeline_name¶

The name of the pipeline definition.

Type

Optional[str]

run_id¶

The id of the run that produced the output.

Type

Optional[str]

metadata¶

A dict of the metadata that is assigned to the OutputDefinition that produced the output.

Type

Optional[Dict[str, Any]]

mapping_key¶

The key that identifies a unique mapped output. None for regular outputs.

Type

Optional[str]

config¶

The configuration for the output.

Type

Optional[Any]

solid_def¶

The definition of the solid that produced the output.

Type

Optional[SolidDefinition]

dagster_type¶

The type of this output.

Type

Optional[DagsterType]

log¶

The log manager to use for this output.

Type

Optional[DagsterLogManager]

version¶

(Experimental) The version of the output.

Type

Optional[str]

resource_config¶

The config associated with the resource that initializes the RootInputManager.

Type

Optional[Dict[str, Any]]

resources¶

The resources required by the output manager, specified by the required_resource_keys parameter.

Type

Optional[Resources]

op_def¶

The definition of the op that produced the output.

Type

Optional[OpDefinition]

add_output_metadata(metadata)[source]¶

Add a dictionary of metadata to the handled output.

Metadata entries added will show up in the HANDLED_OUTPUT and ASSET_MATERIALIZATION events for the run.

Parameters

metadata (Dict[str, Any]) – A metadata dictionary to log

Examples:

from dagster import IOManager

class MyIOManager(IOManager):
    def handle_output(self, context, obj):
        context.add_output_metadata({"foo": "bar"})

property asset_partition_key¶

The partition key for output asset.

Raises an error if the output asset has no partitioning, or if the run covers a partition range for the output asset.

property asset_partition_key_range¶

The partition key range for output asset.

Raises an error if the output asset has no partitioning.

property asset_partitions_time_window¶

The time window for the partitions of the output asset.

Raises an error if either of the following are true: - The output asset has no partitioning. - The output asset is not partitioned with a TimeWindowPartitionsDefinition.

consume_events()[source]¶

Pops and yields all user-generated events that have been recorded from this context.

If consume_events has not yet been called, this will yield all logged events since the call to handle_output. If consume_events has been called, it will yield all events since the last time consume_events was called. Designed for internal use. Users should never need to invoke this method.

consume_logged_metadata_entries()[source]¶

Pops and yields all user-generated metadata entries that have been recorded from this context.

If consume_logged_metadata_entries has not yet been called, this will yield all logged events since the call to handle_output. If consume_logged_metadata_entries has been called, it will yield all events since the last time consume_logged_metadata_entries was called. Designed for internal use. Users should never need to invoke this method.

get_logged_events()[source]¶

Retrieve the list of user-generated events that were logged via the context.

User-generated events that were yielded will not appear in this list.

Examples:

from dagster import IOManager, build_output_context, AssetMaterialization

class MyIOManager(IOManager):
    def handle_output(self, context, obj):
        ...

def test_handle_output():
    mgr = MyIOManager()
    context = build_output_context()
    mgr.handle_output(context)
    all_user_events = context.get_logged_events()
    materializations = [event for event in all_user_events if isinstance(event, AssetMaterialization)]
    ...

get_logged_metadata_entries()[source]¶

Get the list of metadata entries that have been logged for use with this output.

get_output_identifier()[source]¶

Utility method to get a collection of identifiers that as a whole represent a unique step output.

If not using memoization, the unique identifier collection consists of

• run_id: the id of the run which generates the output.

  Note: This method also handles the re-execution memoization logic. If the step that generates the output is skipped in the re-execution, the run_id will be the id of its parent run.

• step_key: the key for a compute step.

• name: the name of the output. (default: ‘result’).

If using memoization, the version corresponding to the step output is used in place of the run_id.

Returns

A list of identifiers, i.e. (run_id or version), step_key, and output_name

Return type

List[str, ..]

get_run_scoped_output_identifier()[source]¶

Utility method to get a collection of identifiers that as a whole represent a unique step output.

The unique identifier collection consists of

• run_id: the id of the run which generates the output.

  Note: This method also handles the re-execution memoization logic. If the step that generates the output is skipped in the re-execution, the run_id will be the id of its parent run.

• step_key: the key for a compute step.

• name: the name of the output. (default: ‘result’).

Returns

A list of identifiers, i.e. run id, step key, and output name

Return type

List[str, ..]

property has_partition_key¶

Whether the current run is a partitioned run

log_event(event)[source]¶

Log an AssetMaterialization or AssetObservation from within the body of an io manager’s handle_output method.

Events logged with this method will appear in the event log.

Parameters

event (Union[AssetMaterialization, Materialization, AssetObservation]) – The event to log.

Examples:

from dagster import IOManager, AssetMaterialization

class MyIOManager(IOManager):
    def handle_output(self, context, obj):
        context.log_event(AssetMaterialization("foo"))

property partition_key¶

The partition key for the current run.

Raises an error if the current run is not a partitioned run.

dagster.build_input_context(name=None, config=None, metadata=None, upstream_output=None, dagster_type=None, resource_config=None, resources=None, op_def=None, step_context=None)[source]¶

Builds input context from provided parameters.

build_input_context can be used as either a function, or a context manager. If resources that are also context managers are provided, then build_input_context must be used as a context manager.

Parameters

• name (Optional[str]) – The name of the input that we’re loading.

• config (Optional[Any]) – The config attached to the input that we’re loading.

• metadata (Optional[Dict[str, Any]]) – A dict of metadata that is assigned to the InputDefinition that we’re loading for.

• upstream_output (Optional[OutputContext]) – Info about the output that produced the object we’re loading.

• dagster_type (Optional[DagsterType]) – The type of this input.

• resource_config (Optional[Dict[str, Any]]) – The resource config to make available from the input context. This usually corresponds to the config provided to the resource that loads the input manager.

• resources (Optional[Dict[str, Any]]) – The resources to make available from the context. For a given key, you can provide either an actual instance of an object, or a resource definition.

• asset_key (Optional[AssetKey]) – The asset key attached to the InputDefinition.

• op_def (Optional[OpDefinition]) – The definition of the op that’s loading the input.

• step_context (Optional[StepExecutionContext]) – For internal use.

Examples

build_input_context()

with build_input_context(resources={"foo": context_manager_resource}) as context:
    do_something

dagster.build_output_context(step_key=None, name=None, metadata=None, run_id=None, mapping_key=None, config=None, dagster_type=None, version=None, resource_config=None, resources=None, solid_def=None, op_def=None)[source]¶

Builds output context from provided parameters.

build_output_context can be used as either a function, or a context manager. If resources that are also context managers are provided, then build_output_context must be used as a context manager.

Parameters

• step_key (Optional[str]) – The step_key for the compute step that produced the output.

• name (Optional[str]) – The name of the output that produced the output.

• metadata (Optional[Dict[str, Any]]) – A dict of the metadata that is assigned to the OutputDefinition that produced the output.

• mapping_key (Optional[str]) – The key that identifies a unique mapped output. None for regular outputs.

• config (Optional[Any]) – The configuration for the output.

• dagster_type (Optional[DagsterType]) – The type of this output.

• version (Optional[str]) – (Experimental) The version of the output.

• resource_config (Optional[Dict[str, Any]]) – The resource config to make available from the input context. This usually corresponds to the config provided to the resource that loads the output manager.

• resources (Optional[Resources]) – The resources to make available from the context. For a given key, you can provide either an actual instance of an object, or a resource definition.

• solid_def (Optional[SolidDefinition]) – The definition of the solid that produced the output.

• op_def (Optional[OpDefinition]) – The definition of the solid that produced the output.

Examples

build_output_context()

with build_output_context(resources={"foo": context_manager_resource}) as context:
    do_something

Built-in IO Managers¶

dagster.mem_io_manager IOManagerDefinition[source]¶

Built-in IO manager that stores and retrieves values in memory.

dagster.fs_io_manager IOManagerDefinition[source]¶

Built-in filesystem IO manager that stores and retrieves values using pickling.

Allows users to specify a base directory where all the step outputs will be stored. By default, step outputs will be stored in the directory specified by local_artifact_storage in your dagster.yaml file (which will be a temporary directory if not explicitly set).

Serializes and deserializes output values using pickling and automatically constructs the filepaths for the assets.

Example usage:

1. Specify a job-level IO manager using the reserved resource key "io_manager", which will set the given IO manager on all ops in a job.

from dagster import fs_io_manager, job, op

@op
def op_a():
    # create df ...
    return df

@op
def op_b(df):
    return df[:5]

@job(
    resource_defs={
        "io_manager": fs_io_manager.configured({"base_path": "/my/base/path"})
    }
)
def job():
    op_b(op_a())

2. Specify IO manager on Out, which allows the user to set different IO managers on different step outputs.

from dagster import fs_io_manager, job, op, Out

@op(out=Out(io_manager_key="my_io_manager"))
def op_a():
    # create df ...
    return df

@op
def op_b(df):
    return df[:5]

@job(resource_defs={"my_io_manager": fs_io_manager})
def job():
    op_b(op_a())

dagster.custom_path_fs_io_manager IOManagerDefinition[source]¶

Built-in IO manager that allows users to custom output file path per output definition.

It requires users to specify a base directory where all the step output will be stored in. It serializes and deserializes output values (assets) using pickling and stores the pickled object in the user-provided file paths.

Example usage:

from dagster import custom_path_fs_io_manager, job, op

@op(out=Out(metadata={"path": "path/to/sample_output"}))
def sample_data(df):
    return df[:5]

my_custom_path_fs_io_manager = custom_path_fs_io_manager.configured(
    {"base_dir": "path/to/basedir"}
)

@job(resource_defs={"io_manager": my_custom_path_fs_io_manager})
def my_job():
    sample_data()

Root Input Managers (Experimental)¶

Root input managers are user-provided objects that specify how to load inputs that aren’t connected to upstream outputs.

@dagster.root_input_manager(config_schema=None, description=None, input_config_schema=None, required_resource_keys=None, version=None)[source]¶

Define a root input manager.

Root input managers load op inputs that aren’t connected to upstream outputs.

The decorated function should accept a InputContext and resource config, and return a loaded object that will be passed into one of the inputs of an op.

The decorator produces an RootInputManagerDefinition.

Parameters

• config_schema (Optional[ConfigSchema]) – The schema for the resource-level config. If not set, Dagster will accept any config provided.

• description (Optional[str]) – A human-readable description of the resource.

• input_config_schema (Optional[ConfigSchema]) – A schema for the input-level config. Each input that uses this input manager can be configured separately using this config. If not set, Dagster will accept any config provided.

• required_resource_keys (Optional[Set[str]]) – Keys for the resources required by the input manager.

• version (Optional[str]) – (Experimental) the version of the input manager definition.

Examples:

from dagster import root_input_manager, op, job, In

@root_input_manager
def csv_loader(_):
    return read_csv("some/path")

@op(ins={"input1": In(root_manager_key="csv_loader_key")})
def my_op(_, input1):
    do_stuff(input1)

@job(resource_defs={"csv_loader_key": csv_loader})
def my_job():
    my_op()

@root_input_manager(config_schema={"base_dir": str})
def csv_loader(context):
    return read_csv(context.resource_config["base_dir"] + "/some/path")

@root_input_manager(input_config_schema={"path": str})
def csv_loader(context):
    return read_csv(context.config["path"])

class dagster.RootInputManager[source]¶

RootInputManagers are used to load inputs to ops at the root of a job.

The easiest way to define an RootInputManager is with the @root_input_manager decorator.

abstract load_input(context)[source]¶

The user-defined read method that loads data given its metadata.

Parameters

context (InputContext) – The context of the step output that produces this asset.

Returns

The data object.

Return type

Any

class dagster.RootInputManagerDefinition(resource_fn=None, config_schema=None, description=None, input_config_schema=None, required_resource_keys=None, version=None)[source]¶

Definition of a root input manager resource.

Root input managers load op inputs that aren’t connected to upstream outputs.

An RootInputManagerDefinition is a ResourceDefinition whose resource_fn returns an RootInputManager.

The easiest way to create an RootInputManagerDefinition is with the @root_input_manager decorator.

property input_config_schema¶

The schema for per-input configuration for inputs that are managed by this input manager