Source code for dagster.core.storage.init

from collections import namedtuple

from dagster import check
from dagster.core.definitions import (
    IntermediateStorageDefinition,
    ModeDefinition,
    PipelineDefinition,
)
from dagster.core.instance import DagsterInstance
from dagster.core.storage.pipeline_run import PipelineRun
from dagster.core.storage.type_storage import TypeStoragePluginRegistry
from dagster.core.system_config.objects import EnvironmentConfig


[docs]class InitIntermediateStorageContext( namedtuple( "InitIntermediateStorageContext", ( "pipeline_def mode_def intermediate_storage_def pipeline_run instance environment_config " "type_storage_plugin_registry resources intermediate_storage_config" ), ) ): """Intermediate storage-specific initialization context. Attributes: pipeline_def (PipelineDefinition): The definition of the pipeline in context. mode_def (ModeDefinition): The definition of the mode in context. intermediate_storage_def (IntermediateStorageDefinition): The definition of the intermediate storage to be constructed. pipeline_run (PipelineRun): The pipeline run in context. instance (DagsterInstance): The instance. environment_config (EnvironmentConfig): The environment config. type_storage_plugin_registry (TypeStoragePluginRegistry): Registry containing custom type storage plugins. resources (Any): Resources available in context. intermediate_storage_config (Dict[str, Any]): The intermediate storage-specific configuration data provided by the environment config. The schema for this data is defined by the ``config_schema`` argument to :py:class:`IntermediateStorageDefinition`. """ def __new__( cls, pipeline_def, mode_def, intermediate_storage_def, pipeline_run, instance, environment_config, type_storage_plugin_registry, resources, intermediate_storage_config, ): return super(InitIntermediateStorageContext, cls).__new__( cls, pipeline_def=check.inst_param(pipeline_def, "pipeline_def", PipelineDefinition), mode_def=check.inst_param(mode_def, "mode_def", ModeDefinition), intermediate_storage_def=check.inst_param( intermediate_storage_def, "intermediate_storage_def", IntermediateStorageDefinition ), pipeline_run=check.inst_param(pipeline_run, "pipeline_run", PipelineRun), instance=check.inst_param(instance, "instance", DagsterInstance), environment_config=check.inst_param( environment_config, "environment_config", EnvironmentConfig ), type_storage_plugin_registry=check.inst_param( type_storage_plugin_registry, "type_storage_plugin_registry", TypeStoragePluginRegistry, ), resources=check.not_none_param(resources, "resources"), intermediate_storage_config=check.dict_param( intermediate_storage_config, intermediate_storage_config, key_type=str ), )