Page MenuHomeElementl
Differential D6164 Diff 30546 python_modules/dagster-graphql/dagster_graphql/implementation/fetch_runs.py

Changeset View

Standalone View

View Options

python_modules/dagster-graphql/dagster_graphql/implementation/fetch_runs.py

from dagster import PipelineDefinition, check from dagster import PipelineDefinition, check
from dagster.config.validate import validate_config from dagster.config.validate import validate_config
from dagster.core.definitions import create_run_config_schema from dagster.core.definitions import create_run_config_schema
from dagster.core.host_representation import PipelineSelector from dagster.core.host_representation import PipelineSelector
from dagster.core.storage.pipeline_run import PipelineRunsFilter from dagster.core.storage.pipeline_run import PipelineRunsFilter
maxAuthorUnsubmitted
Done
Inline Actions

where there are collisions between the native/dagster type and the graphene type, we import the dagster type with the prefix Dagster

max: where there are collisions between the native/dagster type and the graphene type, we import the…
from dagster.core.storage.tags import TagType, get_tag_type from dagster.core.storage.tags import TagType, get_tag_type
from graphql.execution.base import ResolveInfo from graphql.execution.base import ResolveInfo
from .external import ensure_valid_config, get_external_pipeline_or_raise from .external import ensure_valid_config, get_external_pipeline_or_raise
from .utils import UserFacingGraphQLError, capture_dauphin_error from .utils import UserFacingGraphQLError, capture_error
def is_config_valid(pipeline_def, run_config, mode): def is_config_valid(pipeline_def, run_config, mode):
check.str_param(mode, "mode") check.str_param(mode, "mode")
check.inst_param(pipeline_def, "pipeline_def", PipelineDefinition) check.inst_param(pipeline_def, "pipeline_def", PipelineDefinition)
run_config_schema = create_run_config_schema(pipeline_def, mode) run_config_schema = create_run_config_schema(pipeline_def, mode)
validated_config = validate_config(run_config_schema.environment_type, run_config) validated_config = validate_config(run_config_schema.environment_type, run_config)
return validated_config.success return validated_config.success
def get_validated_config(pipeline_def, run_config, mode): def get_validated_config(pipeline_def, run_config, mode):
from dagster_graphql.schema.errors import DauphinPipelineConfigValidationInvalid from ..schema.pipelines.config import GraphenePipelineConfigValidationInvalid
check.str_param(mode, "mode") check.str_param(mode, "mode")
check.inst_param(pipeline_def, "pipeline_def", PipelineDefinition) check.inst_param(pipeline_def, "pipeline_def", PipelineDefinition)
run_config_schema = create_run_config_schema(pipeline_def, mode) run_config_schema = create_run_config_schema(pipeline_def, mode)
validated_config = validate_config(run_config_schema.environment_type, run_config) validated_config = validate_config(run_config_schema.environment_type, run_config)
if not validated_config.success: if not validated_config.success:
raise UserFacingGraphQLError( raise UserFacingGraphQLError(
DauphinPipelineConfigValidationInvalid.for_validation_errors( GraphenePipelineConfigValidationInvalid.for_validation_errors(
pipeline_def.get_external_pipeline(), validated_config.errors pipeline_def.get_external_pipeline(), validated_config.errors
) )
) )
return validated_config return validated_config
def get_run_by_id(graphene_info, run_id): def get_run_by_id(graphene_info, run_id):
from ..schema.errors import GraphenePipelineRunNotFoundError
from ..schema.pipelines.pipeline import GraphenePipelineRun
instance = graphene_info.context.instance instance = graphene_info.context.instance
run = instance.get_run_by_id(run_id) run = instance.get_run_by_id(run_id)
if not run: if not run:
return graphene_info.schema.type_named("PipelineRunNotFoundError")(run_id) return GraphenePipelineRunNotFoundError(run_id)
else: else:
return graphene_info.schema.type_named("PipelineRun")(run) return GraphenePipelineRun(run)
def get_run_tags(graphene_info): def get_run_tags(graphene_info):
from ..schema.tags import GraphenePipelineTagAndValues
instance = graphene_info.context.instance instance = graphene_info.context.instance
return [ return [
graphene_info.schema.type_named("PipelineTagAndValues")(key=key, values=values) GraphenePipelineTagAndValues(key=key, values=values)
for key, values in instance.get_run_tags() for key, values in instance.get_run_tags()
if get_tag_type(key) != TagType.HIDDEN if get_tag_type(key) != TagType.HIDDEN
] ]
@capture_dauphin_error @capture_error
def get_run_group(graphene_info, run_id): def get_run_group(graphene_info, run_id):
from ..schema.errors import GrapheneRunGroupNotFoundError
from ..schema.pipelines.pipeline import GraphenePipelineRun
from ..schema.runs import GrapheneRunGroup
instance = graphene_info.context.instance instance = graphene_info.context.instance
result = instance.get_run_group(run_id) result = instance.get_run_group(run_id)
if result is None: if result is None:
return graphene_info.schema.type_named("RunGroupNotFoundError")(run_id) return GrapheneRunGroupNotFoundError(run_id)
else: else:
root_run_id, run_group = result root_run_id, run_group = result
return graphene_info.schema.type_named("RunGroup")( return GrapheneRunGroup(
root_run_id=root_run_id, root_run_id=root_run_id, runs=[GraphenePipelineRun(run) for run in run_group]
runs=[graphene_info.schema.type_named("PipelineRun")(run) for run in run_group],
) )
def get_runs(graphene_info, filters, cursor=None, limit=None): def get_runs(graphene_info, filters, cursor=None, limit=None):
from ..schema.pipelines.pipeline import GraphenePipelineRun
check.opt_inst_param(filters, "filters", PipelineRunsFilter) check.opt_inst_param(filters, "filters", PipelineRunsFilter)
check.opt_str_param(cursor, "cursor") check.opt_str_param(cursor, "cursor")
check.opt_int_param(limit, "limit") check.opt_int_param(limit, "limit")
instance = graphene_info.context.instance instance = graphene_info.context.instance
runs = [] runs = []
if filters and filters.run_ids and len(filters.run_ids) == 1: if filters and filters.run_ids and len(filters.run_ids) == 1:
run = instance.get_run_by_id(filters.run_ids[0]) run = instance.get_run_by_id(filters.run_ids[0])
if run: if run:
runs = [run] runs = [run]
elif filters and ( elif filters and (
filters.pipeline_name filters.pipeline_name
or filters.tags or filters.tags
or filters.statuses or filters.statuses
or filters.snapshot_id or filters.snapshot_id
or filters.run_ids or filters.run_ids
): ):
runs = instance.get_runs(filters, cursor, limit) runs = instance.get_runs(filters, cursor, limit)
else: else:
runs = instance.get_runs(cursor=cursor, limit=limit) runs = instance.get_runs(cursor=cursor, limit=limit)
return [graphene_info.schema.type_named("PipelineRun")(run) for run in runs] return [GraphenePipelineRun(run) for run in runs]
def get_runs_count(graphene_info, filters): def get_runs_count(graphene_info, filters):
return graphene_info.context.instance.get_runs_count(filters) return graphene_info.context.instance.get_runs_count(filters)
def get_run_groups(graphene_info, filters=None, cursor=None, limit=None): def get_run_groups(graphene_info, filters=None, cursor=None, limit=None):
from ..schema.runs import DauphinRunGroup from ..schema.pipelines.pipeline import GraphenePipelineRun
from ..schema.runs import GrapheneRunGroup
check.opt_inst_param(filters, "filters", PipelineRunsFilter) check.opt_inst_param(filters, "filters", PipelineRunsFilter)
check.opt_str_param(cursor, "cursor") check.opt_str_param(cursor, "cursor")
check.opt_int_param(limit, "limit") check.opt_int_param(limit, "limit")
instance = graphene_info.context.instance instance = graphene_info.context.instance
run_groups = instance.get_run_groups(filters=filters, cursor=cursor, limit=limit) run_groups = instance.get_run_groups(filters=filters, cursor=cursor, limit=limit)
for root_run_id in run_groups: for root_run_id in run_groups:
run_groups[root_run_id]["runs"] = [ run_groups[root_run_id]["runs"] = [
graphene_info.schema.type_named("PipelineRun")(run) GraphenePipelineRun(run) for run in run_groups[root_run_id]["runs"]
for run in run_groups[root_run_id]["runs"]
] ]
return [ return [
DauphinRunGroup(root_run_id=root_run_id, runs=run_group["runs"]) GrapheneRunGroup(root_run_id=root_run_id, runs=run_group["runs"])
for root_run_id, run_group in run_groups.items() for root_run_id, run_group in run_groups.items()
] ]
@capture_dauphin_error @capture_error
def validate_pipeline_config(graphene_info, selector, run_config, mode): def validate_pipeline_config(graphene_info, selector, run_config, mode):
from ..schema.pipelines.config import GraphenePipelineConfigValidationValid
check.inst_param(graphene_info, "graphene_info", ResolveInfo) check.inst_param(graphene_info, "graphene_info", ResolveInfo)
check.inst_param(selector, "selector", PipelineSelector) check.inst_param(selector, "selector", PipelineSelector)
check.opt_str_param(mode, "mode") check.opt_str_param(mode, "mode")
external_pipeline = get_external_pipeline_or_raise(graphene_info, selector) external_pipeline = get_external_pipeline_or_raise(graphene_info, selector)
ensure_valid_config(external_pipeline, mode, run_config) ensure_valid_config(external_pipeline, mode, run_config)
return graphene_info.schema.type_named("PipelineConfigValidationValid")( return GraphenePipelineConfigValidationValid(pipeline_name=external_pipeline.name)
pipeline_name=external_pipeline.name
)
@capture_dauphin_error @capture_error
def get_execution_plan(graphene_info, selector, run_config, mode): def get_execution_plan(graphene_info, selector, run_config, mode):
from ..schema.execution import GrapheneExecutionPlan
check.inst_param(graphene_info, "graphene_info", ResolveInfo) check.inst_param(graphene_info, "graphene_info", ResolveInfo)
check.inst_param(selector, "selector", PipelineSelector) check.inst_param(selector, "selector", PipelineSelector)
check.opt_str_param(mode, "mode") check.opt_str_param(mode, "mode")
external_pipeline = get_external_pipeline_or_raise(graphene_info, selector) external_pipeline = get_external_pipeline_or_raise(graphene_info, selector)
ensure_valid_config(external_pipeline, mode, run_config) ensure_valid_config(external_pipeline, mode, run_config)
return graphene_info.schema.type_named("ExecutionPlan")( return GrapheneExecutionPlan(
graphene_info.context.get_external_execution_plan( graphene_info.context.get_external_execution_plan(
external_pipeline=external_pipeline, external_pipeline=external_pipeline,
mode=mode, mode=mode,
run_config=run_config, run_config=run_config,
step_keys_to_execute=None, step_keys_to_execute=None,
) )
) )
@capture_dauphin_error @capture_error
def get_stats(graphene_info, run_id): def get_stats(graphene_info, run_id):
from ..schema.pipelines.pipeline_run_stats import GraphenePipelineRunStatsSnapshot
stats = graphene_info.context.instance.get_run_stats(run_id) stats = graphene_info.context.instance.get_run_stats(run_id)
stats.id = "stats-{run_id}" stats.id = "stats-{run_id}"
return graphene_info.schema.type_named("PipelineRunStatsSnapshot")(stats) return GraphenePipelineRunStatsSnapshot(stats)
def get_step_stats(graphene_info, run_id, step_keys=None): def get_step_stats(graphene_info, run_id, step_keys=None):
from ..schema.logs.events import GraphenePipelineRunStepStats
step_stats = graphene_info.context.instance.get_run_step_stats(run_id, step_keys) step_stats = graphene_info.context.instance.get_run_step_stats(run_id, step_keys)
return [graphene_info.schema.type_named("PipelineRunStepStats")(stats) for stats in step_stats] return [GraphenePipelineRunStepStats(stats) for stats in step_stats]