Differential D6164 Diff 30381 python_modules/dagster-graphql/dagster_graphql/implementation/utils.py

Changeset View

Standalone View

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

	import sys			import sys
	from collections import namedtuple			from collections import namedtuple

	from dagster import check			from dagster import check
	from dagster.core.host_representation import PipelineSelector			from dagster.core.host_representation import PipelineSelector
	from dagster.utils.error import serializable_error_info_from_exc_info			from dagster.utils.error import serializable_error_info_from_exc_info


	def capture_dauphin_error(fn):			def capture_error(fn):
				maxAuthorUnsubmitted Done Inline Actions rename max: rename
	def _fn(args, *kwargs):			def _fn(args, *kwargs):
	from dagster_graphql.schema.errors import DauphinPythonError			from dagster_graphql.schema.errors import PythonError

	try:			try:
	return fn(args, *kwargs)			return fn(args, *kwargs)
	except UserFacingGraphQLError as de_exception:			except UserFacingGraphQLError as de_exception:
	return de_exception.dauphin_error			return de_exception.error
	except Exception: # pylint: disable=broad-except			except Exception: # pylint: disable=broad-except
	return DauphinPythonError(serializable_error_info_from_exc_info(sys.exc_info()))			return PythonError(serializable_error_info_from_exc_info(sys.exc_info()))

	return _fn			return _fn


	class UserFacingGraphQLError(Exception):			class UserFacingGraphQLError(Exception):
	def __init__(self, dauphin_error):			def __init__(self, error):
	self.dauphin_error = dauphin_error			self.error = error
	message = "[{cls}] {message}".format(			message = "[{cls}] {message}".format(
	cls=dauphin_error.__class__.__name__,			cls=error.__class__.__name__,
	message=dauphin_error.message if hasattr(dauphin_error, "message") else None,			message=error.message if hasattr(error, "message") else None,
	)			)
	super(UserFacingGraphQLError, self).__init__(message)			super(UserFacingGraphQLError, self).__init__(message)


	def pipeline_selector_from_graphql(context, data):			def pipeline_selector_from_graphql(context, data):
	from dagster_graphql.implementation.context import DagsterGraphQLContext			from dagster_graphql.implementation.context import DagsterGraphQLContext

	check.inst_param(context, "context", DagsterGraphQLContext)			check.inst_param(context, "context", DagsterGraphQLContext)
	▲ Show 20 Lines • Show All 54 Lines • Show Last 20 Lines