Differential D6164 Diff 30546 python_modules/dagster-graphql/dagster_graphql/schema/schedules/ticks.py

Changeset View

Standalone View

python_modules/dagster-graphql/dagster_graphql/schema/schedules/ticks.py

				import graphene
	from dagster.core.scheduler.job import JobTickStatus			from dagster.core.scheduler.job import JobTickStatus

				from ..errors import GraphenePythonError
				from ..jobs import GrapheneJobTickStatus


				class GrapheneScheduleTickSuccessData(graphene.ObjectType):
				run = graphene.Field("dagster_graphql.schema.pipelines.pipeline.GraphenePipelineRun")

				class Meta:
				name = "ScheduleTickSuccessData"


				class GrapheneScheduleTickFailureData(graphene.ObjectType):
				error = graphene.NonNull(GraphenePythonError)

				class Meta:
				name = "ScheduleTickFailureData"


	def tick_specific_data_from_dagster_tick(graphene_info, tick):			def tick_specific_data_from_dagster_tick(graphene_info, tick):
				from ..pipelines.pipeline import GraphenePipelineRun

	if tick.status == JobTickStatus.SUCCESS:			if tick.status == JobTickStatus.SUCCESS:
	if tick.run_ids and graphene_info.context.instance.has_run(tick.run_ids[0]):			if tick.run_ids and graphene_info.context.instance.has_run(tick.run_ids[0]):
	return graphene_info.schema.type_named("ScheduleTickSuccessData")(			return GrapheneScheduleTickSuccessData(
	run=graphene_info.schema.type_named("PipelineRun")(			run=GraphenePipelineRun(
	graphene_info.context.instance.get_run_by_id(tick.run_ids[0])			graphene_info.context.instance.get_run_by_id(tick.run_ids[0])
	)			)
	)			)
	return graphene_info.schema.type_named("ScheduleTickSuccessData")(run=None)			return GrapheneScheduleTickSuccessData(run=None)
	elif tick.status == JobTickStatus.FAILURE:			elif tick.status == JobTickStatus.FAILURE:
	error = tick.error			error = tick.error
	return graphene_info.schema.type_named("ScheduleTickFailureData")(error=error)			return GrapheneScheduleTickFailureData(error=error)


				class GrapheneScheduleTickSpecificData(graphene.Union):
				class Meta:
				types = (
				GrapheneScheduleTickSuccessData,
				GrapheneScheduleTickFailureData,
				)
				name = "ScheduleTickSpecificData"


				class GrapheneScheduleTick(graphene.ObjectType):
				tick_id = graphene.NonNull(graphene.String)
				status = graphene.NonNull(GrapheneJobTickStatus)
				timestamp = graphene.NonNull(graphene.Float)
				tick_specific_data = graphene.Field(GrapheneScheduleTickSpecificData)

				class Meta:
				name = "ScheduleTick"