Changeset View
Changeset View
Standalone View
Standalone View
python_modules/dagster/dagster/core/snap/dep_snapshot.py
| from collections import defaultdict, namedtuple | from collections import defaultdict, namedtuple | ||||
| from dagster import check | from dagster import check | ||||
| from dagster.core.definitions import GraphDefinition | |||||
| from dagster.core.definitions.dependency import Solid, SolidInputHandle | from dagster.core.definitions.dependency import Solid, SolidInputHandle | ||||
| from dagster.core.definitions.solid_container import IContainSolids | |||||
| from dagster.serdes import whitelist_for_serdes | from dagster.serdes import whitelist_for_serdes | ||||
| def build_solid_invocation_snap(icontains_solids, solid): | def build_solid_invocation_snap(icontains_solids, solid): | ||||
| check.inst_param(solid, "solid", Solid) | check.inst_param(solid, "solid", Solid) | ||||
| check.inst_param(icontains_solids, "icontains_solids", IContainSolids) | check.inst_param(icontains_solids, "icontains_solids", GraphDefinition) | ||||
| dep_structure = icontains_solids.dependency_structure | dep_structure = icontains_solids.dependency_structure | ||||
| input_def_snaps = [] | input_def_snaps = [] | ||||
| input_to_outputs_map = dep_structure.input_to_upstream_outputs_for_solid(solid.name) | input_to_outputs_map = dep_structure.input_to_upstream_outputs_for_solid(solid.name) | ||||
| for input_def in solid.definition.input_defs: | for input_def in solid.definition.input_defs: | ||||
| input_handle = SolidInputHandle(solid, input_def) | input_handle = SolidInputHandle(solid, input_def) | ||||
| Show All 11 Lines | return SolidInvocationSnap( | ||||
| solid_name=solid.name, | solid_name=solid.name, | ||||
| solid_def_name=solid.definition.name, | solid_def_name=solid.definition.name, | ||||
| tags=solid.tags, | tags=solid.tags, | ||||
| input_dep_snaps=input_def_snaps, | input_dep_snaps=input_def_snaps, | ||||
| ) | ) | ||||
| def build_dep_structure_snapshot_from_icontains_solids(icontains_solids): | def build_dep_structure_snapshot_from_icontains_solids(icontains_solids): | ||||
| check.inst_param(icontains_solids, "icontains_solids", IContainSolids) | check.inst_param(icontains_solids, "icontains_solids", GraphDefinition) | ||||
| return DependencyStructureSnapshot( | return DependencyStructureSnapshot( | ||||
| solid_invocation_snaps=[ | solid_invocation_snaps=[ | ||||
| build_solid_invocation_snap(icontains_solids, solid) | build_solid_invocation_snap(icontains_solids, solid) | ||||
| for solid in icontains_solids.solids | for solid in icontains_solids.solids | ||||
| ] | ] | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 138 Lines • Show Last 20 Lines | |||||