Differential D8675 Diff 40779 python_modules/dagster/dagster/core/definitions/pipeline.py

Changeset View

Standalone View

python_modules/dagster/dagster/core/definitions/pipeline.py

from functools import update_wrapper		from functools import update_wrapper
from typing import TYPE_CHECKING, AbstractSet, Any, Dict, FrozenSet, List, Optional, Set, Union		from typing import (
		TYPE_CHECKING,
		AbstractSet,
		Any,
		Dict,
		FrozenSet,
		Iterable,
		List,
		Optional,
		Set,
		Union,
		)

from dagster import check		from dagster import check
from dagster.core.definitions.policy import RetryPolicy		from dagster.core.definitions.policy import RetryPolicy
from dagster.core.definitions.resource import ResourceDefinition		from dagster.core.definitions.resource import ResourceDefinition
from dagster.core.definitions.solid import NodeDefinition		from dagster.core.definitions.solid import NodeDefinition
from dagster.core.errors import (		from dagster.core.errors import (
DagsterInvalidDefinitionError,		DagsterInvalidDefinitionError,
DagsterInvalidSubsetError,		DagsterInvalidSubsetError,
▲ Show 20 Lines • Show All 329 Lines • ▼ Show 20 Lines	def get_required_resource_defs_for_mode(self, mode: str) -> Dict[str, ResourceDefinition]:
for resource_key, resource in self.get_mode_definition(mode).resource_defs.items()		for resource_key, resource in self.get_mode_definition(mode).resource_defs.items()
if resource_key in self._resource_requirements[mode]		if resource_key in self._resource_requirements[mode]
}		}

@property		@property
def all_solid_defs(self) -> List[NodeDefinition]:		def all_solid_defs(self) -> List[NodeDefinition]:
return list(self._all_node_defs.values())		return list(self._all_node_defs.values())

		# all_solid_defs should be removed in favor of all_node_defs
		@property
		def all_node_defs(self) -> Iterable[NodeDefinition]:
		return self._all_node_defs.values()

@property		@property
def top_level_solid_defs(self) -> List[NodeDefinition]:		def top_level_solid_defs(self) -> List[NodeDefinition]:
return self._current_level_node_defs		return self._current_level_node_defs

def solid_def_named(self, name: str) -> NodeDefinition:		def solid_def_named(self, name: str) -> NodeDefinition:
check.str_param(name, "name")		check.str_param(name, "name")

check.invariant(name in self._all_node_defs, "{} not found".format(name))		check.invariant(name in self._all_node_defs, "{} not found".format(name))
▲ Show 20 Lines • Show All 644 Lines • Show Last 20 Lines