Skip to content

context

SerializableExecutionContext #

Bases: AbstractComputeExecutionContext

The execution context for a papermill_origami run.

Source code in noteable_dagstermill/context.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class SerializableExecutionContext(AbstractComputeExecutionContext):
    """The execution context for a papermill_origami run."""

    def __init__(self, pipeline_tags, op_config, resources, run_id, run, solid_handle) -> None:
        self._pipeline_tags = pipeline_tags
        self._op_config = op_config
        self._resources = resources
        self._run_id = run_id
        self._run = run
        self._solid_handle = solid_handle

    def has_tag(self, key) -> bool:
        return key in self._pipeline_tags

    def get_tag(self, key: str) -> Optional[str]:
        return self._pipeline_tags.get(key)

    @property
    def run_id(self) -> str:
        return self._run_id

    @property
    def solid_handle(self) -> NodeHandle:
        return self._solid_handle

    @property
    def solid_config(self) -> Any:
        return self._op_config

    @property
    def op_handle(self) -> NodeHandle:
        return self.solid_handle

    @property
    def solid(self) -> Node:
        return self._pipeline_def.get_solid(self.solid_handle)

    @property
    def op(self) -> Node:
        return self.solid

    @property
    def solid_def(self) -> SolidDefinition:
        """SolidDefinition: The current solid definition."""
        return self.solid_def

    @property
    def op_def(self) -> OpDefinition:
        """OpDefinition: The current op definition."""
        return cast(
            OpDefinition,
            check.inst(
                self.solid_def,
                OpDefinition,
                "Called op_def on a legacy solid. Use solid_def instead.",
            ),
        )

    @property
    def job_def(self) -> JobDefinition:
        return self._job_def

    @property
    def run(self) -> DagsterRun:
        return self._run

    @property
    def resources(self) -> Any:
        return self._resources

    @property
    def log(self) -> DagsterLogManager:
        return NotImplemented

    @property
    def op_config(self) -> Any:
        return self._op_config

    @classmethod
    def load(cls, path: Union[str, Path]) -> 'SerializableExecutionContext':
        pass

    def dump(self, path: Union[str, Path]) -> None:
        cloudpickle.register_pickle_by_value(sys.modules[__name__])
        with open(path, "wb") as f:
            cloudpickle.dump(self, f)

    def dumps(self) -> bytes:
        cloudpickle.register_pickle_by_value(sys.modules[__name__])
        return cloudpickle.dumps(self)

op_def() property #

OpDefinition: The current op definition.

Source code in noteable_dagstermill/context.py 
58
59
60
61
62
63
64
65
66
67
68
@property
def op_def(self) -> OpDefinition:
    """OpDefinition: The current op definition."""
    return cast(
        OpDefinition,
        check.inst(
            self.solid_def,
            OpDefinition,
            "Called op_def on a legacy solid. Use solid_def instead.",
        ),
    )

solid_def() property #

SolidDefinition: The current solid definition.

Source code in noteable_dagstermill/context.py 
53
54
55
56
@property
def solid_def(self) -> SolidDefinition:
    """SolidDefinition: The current solid definition."""
    return self.solid_def