Differential D8948 Diff 41811 python_modules/dagster/dagster/cli/api.py

Changeset View

Standalone View

python_modules/dagster/dagster/cli/api.py

Show All 11 Lines
from dagster.core.execution.api import create_execution_plan, execute_plan_iterator		from dagster.core.execution.api import create_execution_plan, execute_plan_iterator
from dagster.core.instance import DagsterInstance		from dagster.core.instance import DagsterInstance
from dagster.core.storage.pipeline_run import PipelineRun		from dagster.core.storage.pipeline_run import PipelineRun
from dagster.core.test_utils import mock_system_timezone		from dagster.core.test_utils import mock_system_timezone
from dagster.core.types.loadable_target_origin import LoadableTargetOrigin		from dagster.core.types.loadable_target_origin import LoadableTargetOrigin
from dagster.grpc import DagsterGrpcClient, DagsterGrpcServer		from dagster.grpc import DagsterGrpcClient, DagsterGrpcServer
from dagster.grpc.impl import core_execute_run		from dagster.grpc.impl import core_execute_run
from dagster.grpc.types import ExecuteRunArgs, ExecuteStepArgs		from dagster.grpc.types import ExecuteRunArgs, ExecuteStepArgs
from dagster.serdes import (		from dagster.serdes import deserialize_as, serialize_dagster_namedtuple, whitelist_for_serdes
deserialize_json_to_dagster_namedtuple,
serialize_dagster_namedtuple,
whitelist_for_serdes,
)
from dagster.seven import nullcontext		from dagster.seven import nullcontext
from dagster.utils.hosted_user_process import recon_pipeline_from_origin		from dagster.utils.hosted_user_process import recon_pipeline_from_origin
from dagster.utils.interrupts import capture_interrupts		from dagster.utils.interrupts import capture_interrupts


@whitelist_for_serdes		@whitelist_for_serdes
class ExecuteRunArgsLoadComplete(namedtuple("_ExecuteRunArgsLoadComplete", "")):		class ExecuteRunArgsLoadComplete(namedtuple("_ExecuteRunArgsLoadComplete", "")):
pass		pass
Show All 12 Lines	@api_cli.command(
help=(		help=(
"[INTERNAL] This is an internal utility. Users should generally not invoke this command "		"[INTERNAL] This is an internal utility. Users should generally not invoke this command "
"interactively."		"interactively."
),		),
)		)
@click.argument("input_json", type=click.STRING)		@click.argument("input_json", type=click.STRING)
def execute_run_command(input_json):		def execute_run_command(input_json):
with capture_interrupts():		with capture_interrupts():
args = check.inst(deserialize_json_to_dagster_namedtuple(input_json), ExecuteRunArgs)		args = deserialize_as(input_json, ExecuteRunArgs)
recon_pipeline = recon_pipeline_from_origin(args.pipeline_origin)		recon_pipeline = recon_pipeline_from_origin(args.pipeline_origin)

with (		with (
DagsterInstance.from_ref(args.instance_ref)		DagsterInstance.from_ref(args.instance_ref)
if args.instance_ref		if args.instance_ref
else DagsterInstance.get()		else DagsterInstance.get()
) as instance:		) as instance:
buffer = []		buffer = []
▲ Show 20 Lines • Show All 98 Lines • ▼ Show 20 Lines	help=(
"[INTERNAL] This is an internal utility. Users should generally not invoke this command "		"[INTERNAL] This is an internal utility. Users should generally not invoke this command "
"interactively."		"interactively."
),		),
)		)
@click.argument("input_json", type=click.STRING)		@click.argument("input_json", type=click.STRING)
def execute_step_command(input_json):		def execute_step_command(input_json):
with capture_interrupts():		with capture_interrupts():

args = check.inst(deserialize_json_to_dagster_namedtuple(input_json), ExecuteStepArgs)		args = deserialize_as(input_json, ExecuteStepArgs)

with (		with (
DagsterInstance.from_ref(args.instance_ref)		DagsterInstance.from_ref(args.instance_ref)
if args.instance_ref		if args.instance_ref
else DagsterInstance.get()		else DagsterInstance.get()
) as instance:		) as instance:
pipeline_run = instance.get_run_by_id(args.pipeline_run_id)		pipeline_run = instance.get_run_by_id(args.pipeline_run_id)
check.inst(		check.inst(
▲ Show 20 Lines • Show All 219 Lines • Show Last 20 Lines