Differential D5083 Diff 26697 python_modules/dagster/dagster/cli/pipeline.py

Changeset View

Standalone View

python_modules/dagster/dagster/cli/pipeline.py

from __future__ import print_function		from __future__ import print_function

		import logging
		Lint: Unused Import Inline Actions Unused import logging Lint: Unused Import: Unused import logging
import os		import os
import re		import re
import sys		import sys
import textwrap		import textwrap

import click		import click
import yaml		import yaml
from dagster import PipelineDefinition, check, execute_pipeline		from dagster import PipelineDefinition, check, execute_pipeline
Show All 27 Lines	from dagster.core.host_representation.external_data import (
ExternalPartitionExecutionErrorData,		ExternalPartitionExecutionErrorData,
ExternalPartitionSetExecutionParamData,		ExternalPartitionSetExecutionParamData,
)		)
from dagster.core.host_representation.selector import PipelineSelector		from dagster.core.host_representation.selector import PipelineSelector
from dagster.core.instance import DagsterInstance		from dagster.core.instance import DagsterInstance
from dagster.core.snap import PipelineSnapshot, SolidInvocationSnap		from dagster.core.snap import PipelineSnapshot, SolidInvocationSnap
from dagster.core.snap.execution_plan_snapshot import ExecutionPlanSnapshotErrorData		from dagster.core.snap.execution_plan_snapshot import ExecutionPlanSnapshotErrorData
from dagster.core.storage.pipeline_run import PipelineRun		from dagster.core.storage.pipeline_run import PipelineRun
		from dagster.core.storage.tags import MEMOIZED_RUN_TAG
		Lint: Unused Import Inline Actions Unused MEMOIZED_RUN_TAG imported from dagster.core.storage.tags Lint: Unused Import: Unused MEMOIZED_RUN_TAG imported from dagster.core.storage.tags
from dagster.core.telemetry import log_external_repo_stats, telemetry_wrapper		from dagster.core.telemetry import log_external_repo_stats, telemetry_wrapper
from dagster.core.utils import make_new_backfill_id		from dagster.core.utils import make_new_backfill_id
from dagster.seven import IS_WINDOWS, JSONDecodeError, json		from dagster.seven import IS_WINDOWS, JSONDecodeError, json
from dagster.utils import (		from dagster.utils import (
DEFAULT_WORKSPACE_YAML_FILENAME,		DEFAULT_WORKSPACE_YAML_FILENAME,
delay_interrupts,		delay_interrupts,
is_str,		is_str,
load_yaml_from_glob_list,		load_yaml_from_glob_list,
▲ Show 20 Lines • Show All 411 Lines • ▼ Show 20 Lines	):
check.opt_str_param(preset, "preset")		check.opt_str_param(preset, "preset")
check.opt_dict_param(tags, "tags", key_type=str)		check.opt_dict_param(tags, "tags", key_type=str)
check.opt_list_param(solid_selection, "solid_selection", of_type=str)		check.opt_list_param(solid_selection, "solid_selection", of_type=str)

run_config, mode, tags, solid_selection = _check_execute_external_pipeline_args(		run_config, mode, tags, solid_selection = _check_execute_external_pipeline_args(
external_pipeline, run_config, mode, preset, tags, solid_selection,		external_pipeline, run_config, mode, preset, tags, solid_selection,
)		)

pipeline_name = external_pipeline.name		pipeline_name = external_pipeline.name
pipeline_selector = PipelineSelector(		pipeline_selector = PipelineSelector(
location_name=repo_location.name,		location_name=repo_location.name,
repository_name=external_repo.name,		repository_name=external_repo.name,
pipeline_name=pipeline_name,		pipeline_name=pipeline_name,
solid_selection=solid_selection,		solid_selection=solid_selection,
		alangenfeldUnsubmitted Not Done Inline Actions did you test this? when does it show up? alangenfeld: did you test this? when does it show up?
		cdecarolisAuthorUnsubmitted Done Inline Actions yea I got rid of this. Only kept the `run_lifecycle.py` path due to it being the relevant path for a playground-launched execution. cdecarolis: yea I got rid of this. Only kept the `run_lifecycle.py` path due to it being the relevant path…
)		)

subset_pipeline_result = repo_location.get_subset_external_pipeline_result(pipeline_selector)		subset_pipeline_result = repo_location.get_subset_external_pipeline_result(pipeline_selector)
if subset_pipeline_result.success == False:		if subset_pipeline_result.success == False:
raise DagsterLaunchFailedError(		raise DagsterLaunchFailedError(
"Failed to load external pipeline subset: {error_message}".format(		"Failed to load external pipeline subset: {error_message}".format(
error_message=subset_pipeline_result.error.message		error_message=subset_pipeline_result.error.message
),		),
▲ Show 20 Lines • Show All 451 Lines • Show Last 20 Lines