DagsterDocs

Source code for dagster.core.definitions.reconstructable

import inspect
import os
import sys
from collections import namedtuple
from functools import lru_cache

from dagster import check, seven
from dagster.core.code_pointer import (
    CodePointer,
    CustomPointer,
    FileCodePointer,
    ModuleCodePointer,
    get_python_file_from_target,
)
from dagster.core.errors import DagsterInvalidSubsetError, DagsterInvariantViolationError
from dagster.core.origin import PipelinePythonOrigin, RepositoryPythonOrigin, SchedulePythonOrigin
from dagster.core.selector import parse_solid_selection
from dagster.serdes import pack_value, unpack_value, whitelist_for_serdes
from dagster.utils.backcompat import experimental

from .pipeline_base import IPipeline


def get_ephemeral_repository_name(pipeline_name):
    check.str_param(pipeline_name, "pipeline_name")
    return "__repository__{pipeline_name}".format(pipeline_name=pipeline_name)


@whitelist_for_serdes
class ReconstructableRepository(
    namedtuple("_ReconstructableRepository", "pointer container_image")
):
    def __new__(
        cls,
        pointer,
        container_image=None,
    ):
        return super(ReconstructableRepository, cls).__new__(
            cls,
            pointer=check.inst_param(pointer, "pointer", CodePointer),
            container_image=check.opt_str_param(container_image, "container_image"),
        )

    @lru_cache(maxsize=1)
    def get_definition(self):
        return repository_def_from_pointer(self.pointer)

    def get_reconstructable_pipeline(self, name):
        return ReconstructablePipeline(self, name)

    def get_reconstructable_schedule(self, name):
        return ReconstructableSchedule(self, name)

    @classmethod
    def for_file(cls, file, fn_name, working_directory=None, container_image=None):
        if not working_directory:
            working_directory = os.getcwd()
        return cls(FileCodePointer(file, fn_name, working_directory), container_image)

    @classmethod
    def for_module(cls, module, fn_name, container_image=None):
        return cls(ModuleCodePointer(module, fn_name), container_image)

    def get_cli_args(self):
        return self.pointer.get_cli_args()

    def get_python_origin(self):
        return RepositoryPythonOrigin(
            executable_path=sys.executable,
            code_pointer=self.pointer,
            container_image=self.container_image,
        )

    def get_python_origin_id(self):
        return self.get_python_origin().get_id()


@whitelist_for_serdes
class ReconstructablePipeline(
    namedtuple(
        "_ReconstructablePipeline",
        "repository pipeline_name solid_selection_str solids_to_execute",
    ),
    IPipeline,
):
    def __new__(
        cls,
        repository,
        pipeline_name,
        solid_selection_str=None,
        solids_to_execute=None,
    ):
        check.opt_set_param(solids_to_execute, "solids_to_execute", of_type=str)
        return super(ReconstructablePipeline, cls).__new__(
            cls,
            repository=check.inst_param(repository, "repository", ReconstructableRepository),
            pipeline_name=check.str_param(pipeline_name, "pipeline_name"),
            solid_selection_str=check.opt_str_param(solid_selection_str, "solid_selection_str"),
            solids_to_execute=solids_to_execute,
        )

    @property
    def solid_selection(self):
        return seven.json.loads(self.solid_selection_str) if self.solid_selection_str else None

    @lru_cache(maxsize=1)
    def get_definition(self):
        return (
            self.repository.get_definition()
            .get_pipeline(self.pipeline_name)
            .get_pipeline_subset_def(self.solids_to_execute)
        )

    def _resolve_solid_selection(self, solid_selection):
        # resolve a list of solid selection queries to a frozenset of qualified solid names
        # e.g. ['foo_solid+'] to {'foo_solid', 'bar_solid'}
        check.list_param(solid_selection, "solid_selection", of_type=str)
        solids_to_execute = parse_solid_selection(self.get_definition(), solid_selection)
        if len(solids_to_execute) == 0:
            raise DagsterInvalidSubsetError(
                "No qualified solids to execute found for solid_selection={requested}".format(
                    requested=solid_selection
                )
            )
        return solids_to_execute

    def get_reconstructable_repository(self):
        return self.repository

    def _subset_for_execution(self, solids_to_execute, solid_selection=None):
        if solids_to_execute:
            pipe = ReconstructablePipeline(
                repository=self.repository,
                pipeline_name=self.pipeline_name,
                solid_selection_str=seven.json.dumps(solid_selection) if solid_selection else None,
                solids_to_execute=frozenset(solids_to_execute),
            )
        else:
            pipe = ReconstructablePipeline(
                repository=self.repository,
                pipeline_name=self.pipeline_name,
            )

        return pipe

    def subset_for_execution(self, solid_selection):
        # take a list of solid queries and resolve the queries to names of solids to execute
        check.opt_list_param(solid_selection, "solid_selection", of_type=str)
        solids_to_execute = (
            self._resolve_solid_selection(solid_selection) if solid_selection else None
        )

        return self._subset_for_execution(solids_to_execute, solid_selection)

    def subset_for_execution_from_existing_pipeline(self, solids_to_execute):
        # take a frozenset of resolved solid names from an existing pipeline
        # so there's no need to parse the selection
        check.opt_set_param(solids_to_execute, "solids_to_execute", of_type=str)

        return self._subset_for_execution(solids_to_execute)

    def describe(self):
        return '"{name}" in repository ({repo})'.format(
            repo=self.repository.pointer.describe, name=self.pipeline_name
        )

    @staticmethod
    def for_file(python_file, fn_name):
        return bootstrap_standalone_recon_pipeline(
            FileCodePointer(python_file, fn_name, os.getcwd())
        )

    @staticmethod
    def for_module(module, fn_name):
        return bootstrap_standalone_recon_pipeline(ModuleCodePointer(module, fn_name))

    def to_dict(self):
        return pack_value(self)

    @staticmethod
    def from_dict(val):
        check.dict_param(val, "val")

        inst = unpack_value(val)
        check.invariant(
            isinstance(inst, ReconstructablePipeline),
            "Deserialized object is not instance of ReconstructablePipeline, got {type}".format(
                type=type(inst)
            ),
        )
        return inst

    def get_python_origin(self):
        return PipelinePythonOrigin(self.pipeline_name, self.repository.get_python_origin())

    def get_python_origin_id(self):
        return self.get_python_origin().get_id()


@whitelist_for_serdes
class ReconstructableSchedule(
    namedtuple(
        "_ReconstructableSchedule",
        "repository schedule_name",
    )
):
    def __new__(
        cls,
        repository,
        schedule_name,
    ):
        return super(ReconstructableSchedule, cls).__new__(
            cls,
            repository=check.inst_param(repository, "repository", ReconstructableRepository),
            schedule_name=check.str_param(schedule_name, "schedule_name"),
        )

    def get_python_origin(self):
        return SchedulePythonOrigin(self.schedule_name, self.repository.get_python_origin())

    def get_python_origin_id(self):
        return self.get_python_origin().get_id()

    @lru_cache(maxsize=1)
    def get_definition(self):
        return self.repository.get_definition().get_schedule_def(self.schedule_name)


[docs]def reconstructable(target): """ Create a :py:class:`~dagster.core.definitions.reconstructable.ReconstructablePipeline` from a function that returns a :py:class:`~dagster.PipelineDefinition`, or a function decorated with :py:func:`@pipeline <dagster.pipeline>` When your pipeline must cross process boundaries, e.g., for execution on multiple nodes or in different systems (like ``dagstermill``), Dagster must know how to reconstruct the pipeline on the other side of the process boundary. This function implements a very conservative strategy for reconstructing pipelines, so that its behavior is easy to predict, but as a consequence it is not able to reconstruct certain kinds of pipelines, such as those defined by lambdas, in nested scopes (e.g., dynamically within a method call), or in interactive environments such as the Python REPL or Jupyter notebooks. If you need to reconstruct pipelines constructed in these ways, you should use :py:func:`~dagster.core.definitions.reconstructable.build_reconstructable_pipeline` instead, which allows you to specify your own strategy for reconstructing a pipeline. Examples: .. code-block:: python from dagster import PipelineDefinition, pipeline, reconstructable @pipeline def foo_pipeline(): ... reconstructable_foo_pipeline = reconstructable(foo_pipeline) def make_bar_pipeline(): return PipelineDefinition(...) reconstructable_bar_pipeline = reconstructable(bar_pipeline) """ from dagster.core.definitions import PipelineDefinition if not seven.is_function_or_decorator_instance_of(target, PipelineDefinition): raise DagsterInvariantViolationError( "Reconstructable target should be a function or definition produced " "by a decorated function, got {type}.".format(type=type(target)), ) if seven.is_lambda(target): raise DagsterInvariantViolationError( "Reconstructable target can not be a lambda. Use a function or " "decorated function defined at module scope instead, or use " "build_reconstructable_pipeline." ) if seven.qualname_differs(target): raise DagsterInvariantViolationError( 'Reconstructable target "{target.__name__}" has a different ' '__qualname__ "{target.__qualname__}" indicating it is not ' "defined at module scope. Use a function or decorated function " "defined at module scope instead, or use build_reconstructable_pipeline.".format( target=target ) ) try: if ( hasattr(target, "__module__") and hasattr(target, "__name__") and inspect.getmodule(target).__name__ != "__main__" ): return ReconstructablePipeline.for_module(target.__module__, target.__name__) except: # pylint: disable=bare-except pass python_file = get_python_file_from_target(target) if not python_file: raise DagsterInvariantViolationError( "reconstructable() can not reconstruct pipelines defined in interactive environments " "like <stdin>, IPython, or Jupyter notebooks. " "Use a pipeline defined in a module or file instead, or " "use build_reconstructable_pipeline." ) pointer = FileCodePointer( python_file=python_file, fn_name=target.__name__, working_directory=os.getcwd() ) return bootstrap_standalone_recon_pipeline(pointer)
[docs]@experimental def build_reconstructable_pipeline( reconstructor_module_name, reconstructor_function_name, reconstructable_args=None, reconstructable_kwargs=None, ): """ Create a :py:class:`dagster.core.definitions.reconstructable.ReconstructablePipeline`. When your pipeline must cross process boundaries, e.g., for execution on multiple nodes or in different systems (like ``dagstermill``), Dagster must know how to reconstruct the pipeline on the other side of the process boundary. This function allows you to use the strategy of your choice for reconstructing pipelines, so that you can reconstruct certain kinds of pipelines that are not supported by :py:func:`~dagster.reconstructable`, such as those defined by lambdas, in nested scopes (e.g., dynamically within a method call), or in interactive environments such as the Python REPL or Jupyter notebooks. If you need to reconstruct pipelines constructed in these ways, use this function instead of :py:func:`~dagster.reconstructable`. Args: reconstructor_module_name (str): The name of the module containing the function to use to reconstruct the pipeline. reconstructor_function_name (str): The name of the function to use to reconstruct the pipeline. reconstructable_args (Tuple): Args to the function to use to reconstruct the pipeline. Values of the tuple must be JSON serializable. reconstructable_kwargs (Dict[str, Any]): Kwargs to the function to use to reconstruct the pipeline. Values of the dict must be JSON serializable. Examples: .. code-block:: python # module: mymodule from dagster import PipelineDefinition, pipeline, build_reconstructable_pipeline class PipelineFactory: def make_pipeline(*args, **kwargs): @pipeline def _pipeline(...): ... return _pipeline def reconstruct_pipeline(*args): factory = PipelineFactory() return factory.make_pipeline(*args) factory = PipelineFactory() foo_pipeline_args = (...,...) foo_pipeline_kwargs = {...:...} foo_pipeline = factory.make_pipeline(*foo_pipeline_args, **foo_pipeline_kwargs) reconstructable_foo_pipeline = build_reconstructable_pipeline( 'mymodule', 'reconstruct_pipeline', foo_pipeline_args, foo_pipeline_kwargs, ) """ check.str_param(reconstructor_module_name, "reconstructor_module_name") check.str_param(reconstructor_function_name, "reconstructor_function_name") reconstructable_args = list(check.opt_tuple_param(reconstructable_args, "reconstructable_args")) reconstructable_kwargs = list( ( [key, value] for key, value in check.opt_dict_param( reconstructable_kwargs, "reconstructable_kwargs", key_type=str ).items() ) ) reconstructor_pointer = ModuleCodePointer( reconstructor_module_name, reconstructor_function_name ) pointer = CustomPointer(reconstructor_pointer, reconstructable_args, reconstructable_kwargs) pipeline_def = pipeline_def_from_pointer(pointer) return ReconstructablePipeline( repository=ReconstructableRepository(pointer), # creates ephemeral repo pipeline_name=pipeline_def.name, )
def bootstrap_standalone_recon_pipeline(pointer): # So this actually straps the the pipeline for the sole # purpose of getting the pipeline name. If we changed ReconstructablePipeline # to get the pipeline on demand in order to get name, we could avoid this. pipeline_def = pipeline_def_from_pointer(pointer) return ReconstructablePipeline( repository=ReconstructableRepository(pointer), # creates ephemeral repo pipeline_name=pipeline_def.name, ) def _check_is_loadable(definition): from .pipeline import PipelineDefinition from .repository import RepositoryDefinition if not isinstance(definition, (PipelineDefinition, RepositoryDefinition)): raise DagsterInvariantViolationError( ( "Loadable attributes must be either a PipelineDefinition or a " "RepositoryDefinition. Got {definition}." ).format(definition=repr(definition)) ) return definition def load_def_in_module(module_name, attribute): return def_from_pointer(CodePointer.from_module(module_name, attribute)) def load_def_in_package(package_name, attribute): return def_from_pointer(CodePointer.from_python_package(package_name, attribute)) def load_def_in_python_file(python_file, attribute, working_directory): return def_from_pointer(CodePointer.from_python_file(python_file, attribute, working_directory)) def def_from_pointer(pointer): target = pointer.load_target() from .pipeline import PipelineDefinition from .repository import RepositoryDefinition if isinstance(target, (PipelineDefinition, RepositoryDefinition)) or not callable(target): return _check_is_loadable(target) # if its a function invoke it - otherwise we are pointing to a # artifact in module scope, likely decorator output if seven.get_args(target): raise DagsterInvariantViolationError( "Error invoking function at {target} with no arguments. " "Reconstructable target must be callable with no arguments".format( target=pointer.describe() ) ) return _check_is_loadable(target()) def pipeline_def_from_pointer(pointer): from .pipeline import PipelineDefinition target = def_from_pointer(pointer) if isinstance(target, PipelineDefinition): return target raise DagsterInvariantViolationError( "CodePointer ({str}) must resolve to a PipelineDefinition. " "Received a {type}".format(str=pointer.describe(), type=type(target)) ) def repository_def_from_target_def(target): from .pipeline import PipelineDefinition from .repository import RepositoryData, RepositoryDefinition # special case - we can wrap a single pipeline in a repository if isinstance(target, PipelineDefinition): # consider including pipeline name in generated repo name return RepositoryDefinition( name=get_ephemeral_repository_name(target.name), repository_data=RepositoryData.from_list([target]), ) elif isinstance(target, RepositoryDefinition): return target else: return None def repository_def_from_pointer(pointer): target = def_from_pointer(pointer) repo_def = repository_def_from_target_def(target) if not repo_def: raise DagsterInvariantViolationError( "CodePointer ({str}) must resolve to a " "RepositoryDefinition or a PipelineDefinition. " "Received a {type}".format(str=pointer.describe(), type=type(target)) ) return repo_def