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

Changeset View

Standalone View

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

				import graphene
	from dagster import check			from dagster import check
	from dagster.core.host_representation import ExternalSchedule, ScheduleSelector			from dagster.core.host_representation import ExternalSchedule, ScheduleSelector
	from dagster.core.host_representation.selector import RepositorySelector			from dagster.core.host_representation.selector import RepositorySelector
	from dagster.core.scheduler.job import JobTickStatsSnapshot, JobTickStatus			from dagster.core.scheduler.job import JobTickStatsSnapshot
	from dagster.core.storage.pipeline_run import PipelineRunsFilter
	from dagster_graphql import dauphin			from ...implementation.fetch_schedules import (
	from dagster_graphql.implementation.fetch_schedules import (
	reconcile_scheduler_state,			reconcile_scheduler_state,
	start_schedule,			start_schedule,
	stop_schedule,			stop_schedule,
	)			)
	from dagster_graphql.schema.errors import (			from ..errors import (
	DauphinPythonError,			GraphenePythonError,
	DauphinRepositoryNotFoundError,			GrapheneRepositoryNotFoundError,
	DauphinScheduleNotFoundError,			GrapheneScheduleNotFoundError,
	DauphinSchedulerNotDefinedError,			GrapheneSchedulerNotDefinedError,
	)			)
				from ..inputs import GrapheneRepositorySelector, GrapheneScheduleSelector
				from ..jobs import GrapheneJobState
	from .schedules import (			from .schedules import (
	DauphinSchedule,			GrapheneSchedule,
	DauphinScheduleOrError,			GrapheneScheduleOrError,
	DauphinSchedules,			GrapheneSchedules,
	DauphinSchedulesOrError,			GrapheneSchedulesOrError,
	)			)
				from .ticks import GrapheneJobTickStatus


	class DauphinScheduleStatus(dauphin.Enum):			class GrapheneScheduleStatus(graphene.Enum):
	class Meta:
	name = "ScheduleStatus"

	RUNNING = "RUNNING"			RUNNING = "RUNNING"
	STOPPED = "STOPPED"			STOPPED = "STOPPED"
	ENDED = "ENDED"			ENDED = "ENDED"


	class DauphinSchedulerOrError(dauphin.Union):
	class Meta:			class Meta:
	name = "SchedulerOrError"			name = "ScheduleStatus"
	types = ("Scheduler", DauphinSchedulerNotDefinedError, "PythonError")


	DauphinJobTickStatus = dauphin.Enum.from_enum(JobTickStatus)


	class DauphinScheduleTick(dauphin.ObjectType):
	class Meta:
	name = "ScheduleTick"

	tick_id = dauphin.NonNull(dauphin.String)
	status = dauphin.NonNull("JobTickStatus")
	timestamp = dauphin.NonNull(dauphin.Float)
	tick_specific_data = dauphin.Field("ScheduleTickSpecificData")

				class GrapheneScheduler(graphene.ObjectType):
				scheduler_class = graphene.String()

	class DauphinScheduleTickSuccessData(dauphin.ObjectType):
	class Meta:			class Meta:
	name = "ScheduleTickSuccessData"			name = "Scheduler"

	run = dauphin.Field("PipelineRun")


	class DauphinScheduleTickFailureData(dauphin.ObjectType):			class GrapheneSchedulerOrError(graphene.Union):
	class Meta:			class Meta:
	name = "ScheduleTickFailureData"			types = (GrapheneScheduler, GrapheneSchedulerNotDefinedError, GraphenePythonError)
				name = "SchedulerOrError"
	error = dauphin.NonNull("PythonError")


	class DauphinScheduleTickSpecificData(dauphin.Union):			class GrapheneScheduleTickStatsSnapshot(graphene.ObjectType):
	class Meta:			ticks_started = graphene.NonNull(graphene.Int)
	name = "ScheduleTickSpecificData"			ticks_succeeded = graphene.NonNull(graphene.Int)
	types = (			ticks_skipped = graphene.NonNull(graphene.Int)
	DauphinScheduleTickSuccessData,			ticks_failed = graphene.NonNull(graphene.Int)
	DauphinScheduleTickFailureData,
	)


	class DauphinScheduleTickStatsSnapshot(dauphin.ObjectType):
	class Meta:			class Meta:
	name = "ScheduleTickStatsSnapshot"			name = "ScheduleTickStatsSnapshot"

	ticks_started = dauphin.NonNull(dauphin.Int)
	ticks_succeeded = dauphin.NonNull(dauphin.Int)
	ticks_skipped = dauphin.NonNull(dauphin.Int)
	ticks_failed = dauphin.NonNull(dauphin.Int)

	def __init__(self, stats):			def __init__(self, stats):
	super(DauphinScheduleTickStatsSnapshot, self).__init__(			super().__init__(
	ticks_started=stats.ticks_started,			ticks_started=stats.ticks_started,
	ticks_succeeded=stats.ticks_succeeded,			ticks_succeeded=stats.ticks_succeeded,
	ticks_skipped=stats.ticks_skipped,			ticks_skipped=stats.ticks_skipped,
	ticks_failed=stats.ticks_failed,			ticks_failed=stats.ticks_failed,
	)			)
	self._stats = check.inst_param(stats, "stats", JobTickStatsSnapshot)			self._stats = check.inst_param(stats, "stats", JobTickStatsSnapshot)


	class DauphinScheduler(dauphin.ObjectType):			class GrapheneReconcileSchedulerStateSuccess(graphene.ObjectType):
	class Meta:			message = graphene.NonNull(graphene.String)
	name = "Scheduler"

	scheduler_class = dauphin.String()


	class DauphinReconcileSchedulerStateSuccess(dauphin.ObjectType):
	class Meta:			class Meta:
	name = "ReconcileSchedulerStateSuccess"			name = "ReconcileSchedulerStateSuccess"

	message = dauphin.NonNull(dauphin.String)


	class DauphinReconcilScheduleStateMutationResult(dauphin.Union):			class GrapheneReconcileSchedulerStateMutationResult(graphene.Union):
	class Meta:			class Meta:
				types = (GraphenePythonError, GrapheneReconcileSchedulerStateSuccess)
	name = "ReconcileSchedulerStateMutationResult"			name = "ReconcileSchedulerStateMutationResult"
	types = (DauphinPythonError, DauphinReconcileSchedulerStateSuccess)


	class DauphinReconcileSchedulerStateMutation(dauphin.Mutation):			class GrapheneReconcileSchedulerStateMutation(graphene.Mutation):
	class Meta:			Output = graphene.NonNull(GrapheneReconcileSchedulerStateMutationResult)
	name = "ReconcileSchedulerStateMutation"

	class Arguments:			class Arguments:
	repository_selector = dauphin.NonNull("RepositorySelector")			repository_selector = graphene.NonNull(GrapheneRepositorySelector)

	Output = dauphin.NonNull("ReconcileSchedulerStateMutationResult")			class Meta:
				name = "ReconcileSchedulerStateMutation"

	def mutate(self, graphene_info, repository_selector):			def mutate(self, graphene_info, repository_selector):
	return reconcile_scheduler_state(			return reconcile_scheduler_state(
	graphene_info, RepositorySelector.from_graphql_input(repository_selector)			graphene_info, RepositorySelector.from_graphql_input(repository_selector)
	)			)


	class DauphinScheduleStateResult(dauphin.ObjectType):			class GrapheneScheduleStateResult(graphene.ObjectType):
				scheduleState = graphene.NonNull(GrapheneJobState)

	class Meta:			class Meta:
	name = "ScheduleStateResult"			name = "ScheduleStateResult"

	scheduleState = dauphin.NonNull("JobState")


	class DauphinScheduleMutationResult(dauphin.Union):			class GrapheneScheduleMutationResult(graphene.Union):
	class Meta:			class Meta:
				types = (GraphenePythonError, GrapheneScheduleStateResult)
	name = "ScheduleMutationResult"			name = "ScheduleMutationResult"
	types = (DauphinPythonError, DauphinScheduleStateResult)


	class DauphinStartScheduleMutation(dauphin.Mutation):			class GrapheneStartScheduleMutation(graphene.Mutation):
	class Meta:			Output = graphene.NonNull(GrapheneScheduleMutationResult)
	name = "StartScheduleMutation"

	class Arguments:			class Arguments:
	schedule_selector = dauphin.NonNull("ScheduleSelector")			schedule_selector = graphene.NonNull(GrapheneScheduleSelector)

	Output = dauphin.NonNull("ScheduleMutationResult")			class Meta:
				name = "StartScheduleMutation"

	def mutate(self, graphene_info, schedule_selector):			def mutate(self, graphene_info, schedule_selector):
	return start_schedule(graphene_info, ScheduleSelector.from_graphql_input(schedule_selector))			return start_schedule(graphene_info, ScheduleSelector.from_graphql_input(schedule_selector))


	class DauphinStopRunningScheduleMutation(dauphin.Mutation):			class GrapheneStopRunningScheduleMutation(graphene.Mutation):
	class Meta:			Output = graphene.NonNull(GrapheneScheduleMutationResult)
	name = "StopRunningScheduleMutation"

	class Arguments:			class Arguments:
	schedule_origin_id = dauphin.NonNull(dauphin.String)			schedule_origin_id = graphene.NonNull(graphene.String)

	Output = dauphin.NonNull("ScheduleMutationResult")			class Meta:
				name = "StopRunningScheduleMutation"

	def mutate(self, graphene_info, schedule_origin_id):			def mutate(self, graphene_info, schedule_origin_id):
	return stop_schedule(graphene_info, schedule_origin_id)			return stop_schedule(graphene_info, schedule_origin_id)


				def types():
				from .ticks import (
				GrapheneScheduleTick,
				GrapheneScheduleTickFailureData,
				GrapheneScheduleTickSpecificData,
				GrapheneScheduleTickSuccessData,
				)

				# Double check mutations don't appear twice
				return [
				GrapheneJobTickStatus,
				GrapheneReconcileSchedulerStateMutation,
				GrapheneReconcileSchedulerStateMutationResult,
				GrapheneReconcileSchedulerStateSuccess,
				GrapheneSchedule,
				GrapheneScheduleMutationResult,
				GrapheneScheduleOrError,
				GrapheneScheduler,
				GrapheneSchedulerOrError,
				GrapheneSchedules,
				GrapheneSchedulesOrError,
				GrapheneScheduleStateResult,
				GrapheneScheduleStatus,
				GrapheneScheduleTick,
				GrapheneScheduleTickFailureData,
				GrapheneScheduleTickSpecificData,
				GrapheneScheduleTickStatsSnapshot,
				GrapheneScheduleTickSuccessData,
				GrapheneStartScheduleMutation,
				GrapheneStopRunningScheduleMutation,
				]