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

Changeset View

Standalone View

python_modules/dagster-graphql/dagster_graphql/schema/config_types.py

				import graphene
	from dagster import check			from dagster import check
	from dagster.config.config_type import ConfigTypeKind			from dagster.config.config_type import ConfigTypeKind
	from dagster.config.snap import get_recursive_type_keys			from dagster.config.snap import get_recursive_type_keys
	from dagster.core.snap import ConfigFieldSnap, ConfigSchemaSnapshot, ConfigTypeSnap			from dagster.core.snap import ConfigFieldSnap, ConfigSchemaSnapshot, ConfigTypeSnap
	from dagster_graphql import dauphin

				from .util import non_null_list

	def to_dauphin_config_type(config_schema_snapshot, config_type_key):
				def to_config_type(config_schema_snapshot, config_type_key):
	check.inst_param(config_schema_snapshot, "config_schema_snapshot", ConfigSchemaSnapshot)			check.inst_param(config_schema_snapshot, "config_schema_snapshot", ConfigSchemaSnapshot)
	check.str_param(config_type_key, "config_type_key")			check.str_param(config_type_key, "config_type_key")

	config_type_snap = config_schema_snapshot.get_config_snap(config_type_key)			config_type_snap = config_schema_snapshot.get_config_snap(config_type_key)
	kind = config_type_snap.kind			kind = config_type_snap.kind

	if kind == ConfigTypeKind.ENUM:			if kind == ConfigTypeKind.ENUM:
	return DauphinEnumConfigType(config_schema_snapshot, config_type_snap)			return GrapheneEnumConfigType(config_schema_snapshot, config_type_snap)
	elif ConfigTypeKind.has_fields(kind):			elif ConfigTypeKind.has_fields(kind):
	return DauphinCompositeConfigType(config_schema_snapshot, config_type_snap)			return GrapheneCompositeConfigType(config_schema_snapshot, config_type_snap)
	elif kind == ConfigTypeKind.ARRAY:			elif kind == ConfigTypeKind.ARRAY:
	return DauphinArrayConfigType(config_schema_snapshot, config_type_snap)			return GrapheneArrayConfigType(config_schema_snapshot, config_type_snap)
	elif kind == ConfigTypeKind.NONEABLE:			elif kind == ConfigTypeKind.NONEABLE:
	return DauphinNullableConfigType(config_schema_snapshot, config_type_snap)			return GrapheneNullableConfigType(config_schema_snapshot, config_type_snap)
	elif kind == ConfigTypeKind.ANY or kind == ConfigTypeKind.SCALAR:			elif kind == ConfigTypeKind.ANY or kind == ConfigTypeKind.SCALAR:
	return DauphinRegularConfigType(config_schema_snapshot, config_type_snap)			return GrapheneRegularConfigType(config_schema_snapshot, config_type_snap)
	elif kind == ConfigTypeKind.SCALAR_UNION:			elif kind == ConfigTypeKind.SCALAR_UNION:
	return DauphinScalarUnionConfigType(config_schema_snapshot, config_type_snap)			return GrapheneScalarUnionConfigType(config_schema_snapshot, config_type_snap)
	else:			else:
	check.failed("Should never reach")			check.failed("Should never reach")


	def _ctor_kwargs_for_snap(config_type_snap):			def _ctor_kwargs_for_snap(config_type_snap):
	return dict(			return dict(
	key=config_type_snap.key,			key=config_type_snap.key,
	description=config_type_snap.description,			description=config_type_snap.description,
	is_selector=config_type_snap.kind == ConfigTypeKind.SELECTOR,			is_selector=config_type_snap.kind == ConfigTypeKind.SELECTOR,
	type_param_keys=config_type_snap.type_param_keys or [],			type_param_keys=config_type_snap.type_param_keys or [],
	)			)


	class DauphinConfigType(dauphin.Interface):			class GrapheneConfigType(graphene.Interface):
	class Meta:			key = graphene.NonNull(graphene.String)
	maxAuthorUnsubmitted Done Inline Actions drop this boilerplate max: drop this boilerplate
	name = "ConfigType"			description = graphene.String()

	key = dauphin.NonNull(dauphin.String)
	description = dauphin.String()

	recursive_config_types = dauphin.Field(			recursive_config_types = graphene.Field(
	dauphin.non_null_list("ConfigType"),			non_null_list(lambda: GrapheneConfigType),
				maxAuthorUnsubmitted Done Inline Actions wrap in lambda to avoid circular reference issue max: wrap in lambda to avoid circular reference issue
	description="""			description="""
	This is an odd and problematic field. It recursively goes down to			This is an odd and problematic field. It recursively goes down to
	get all the types contained within a type. The case where it is horrible			get all the types contained within a type. The case where it is horrible
	are dictionaries and it recurses all the way down to the leaves. This means			are dictionaries and it recurses all the way down to the leaves. This means
	that in a case where one is fetching all the types and then all the inner			that in a case where one is fetching all the types and then all the inner
	types keys for those types, we are returning O(N^2) type keys, which			types keys for those types, we are returning O(N^2) type keys, which
	can cause awful performance for large schemas. When you have access			can cause awful performance for large schemas. When you have access
	to all the types, you should instead only use the type_param_keys			to all the types, you should instead only use the type_param_keys
	field for closed generic types and manually navigate down the to			field for closed generic types and manually navigate down the to
	field types client-side.			field types client-side.

	Where it is useful is when you are fetching types independently and			Where it is useful is when you are fetching types independently and
	want to be able to render them, but without fetching the entire schema.			want to be able to render them, but without fetching the entire schema.

	We use this capability when rendering the sidebar.			We use this capability when rendering the sidebar.
	""",			""",
	)			)
	type_param_keys = dauphin.Field(			type_param_keys = graphene.Field(
	dauphin.non_null_list(dauphin.String),			non_null_list(graphene.String),
	description="""			description="""
	This returns the keys for type parameters of any closed generic type,			This returns the keys for type parameters of any closed generic type,
	(e.g. List, Optional). This should be used for reconstructing and			(e.g. List, Optional). This should be used for reconstructing and
	navigating the full schema client-side and not innerTypes.			navigating the full schema client-side and not innerTypes.
	""",			""",
	)			)
	is_selector = dauphin.NonNull(dauphin.Boolean)			is_selector = graphene.NonNull(graphene.Boolean)

				class Meta:
				name = "ConfigType"


	class ConfigTypeMixin:			class ConfigTypeMixin:
	def __init__(self, config_schema_snapshot, config_type_snap):			def __init__(self, config_schema_snapshot, config_type_snap):
	self._config_type_snap = check.inst_param(			self._config_type_snap = check.inst_param(
	config_type_snap, "config_type_snap", ConfigTypeSnap			config_type_snap, "config_type_snap", ConfigTypeSnap
	)			)
	self._config_schema_snapshot = check.inst_param(			self._config_schema_snapshot = check.inst_param(
	config_schema_snapshot, "config_schema_snapshot", ConfigSchemaSnapshot			config_schema_snapshot, "config_schema_snapshot", ConfigSchemaSnapshot
	)			)
	super(ConfigTypeMixin, self).__init__(**_ctor_kwargs_for_snap(config_type_snap))			super().__init__(**_ctor_kwargs_for_snap(config_type_snap))

	def resolve_recursive_config_types(self, _graphene_info):			def resolve_recursive_config_types(self, _graphene_info):
	return list(			return list(
	map(			map(
	lambda key: to_dauphin_config_type(self._config_schema_snapshot, key),			lambda key: to_config_type(self._config_schema_snapshot, key),
	get_recursive_type_keys(self._config_type_snap, self._config_schema_snapshot),			get_recursive_type_keys(self._config_type_snap, self._config_schema_snapshot),
	)			)
	)			)


	class DauphinRegularConfigType(ConfigTypeMixin, dauphin.ObjectType):			class GrapheneRegularConfigType(ConfigTypeMixin, graphene.ObjectType):
	class Meta:			class Meta:
	name = "RegularConfigType"			interfaces = (GrapheneConfigType,)
				maxAuthorUnsubmitted Done Inline Actions move all interfaces to use tuples rther than lists max: move all interfaces to use tuples rther than lists
	interfaces = [DauphinConfigType]
	description = "Regular is an odd name in this context. It really means Scalar or Any."			description = "Regular is an odd name in this context. It really means Scalar or Any."
				name = "RegularConfigType"

	given_name = dauphin.NonNull(dauphin.String)			given_name = graphene.NonNull(graphene.String)

	def resolve_given_name(self, _):			def resolve_given_name(self, _):
	return self._config_type_snap.given_name			return self._config_type_snap.given_name


	class DauphinWrappingConfigType(dauphin.Interface):			class GrapheneWrappingConfigType(graphene.Interface):
				of_type = graphene.Field(graphene.NonNull(GrapheneConfigType))

	class Meta:			class Meta:
	name = "WrappingConfigType"			name = "WrappingConfigType"

	of_type = dauphin.Field(dauphin.NonNull(DauphinConfigType))

				class GrapheneArrayConfigType(ConfigTypeMixin, graphene.ObjectType):
	class DauphinArrayConfigType(ConfigTypeMixin, dauphin.ObjectType):
	class Meta:			class Meta:
				interfaces = (GrapheneConfigType, GrapheneWrappingConfigType)
	name = "ArrayConfigType"			name = "ArrayConfigType"
	interfaces = [DauphinConfigType, DauphinWrappingConfigType]

	def resolve_of_type(self, _graphene_info):			def resolve_of_type(self, _graphene_info):
	return to_dauphin_config_type(			return to_config_type(self._config_schema_snapshot, self._config_type_snap.inner_type_key,)
	self._config_schema_snapshot, self._config_type_snap.inner_type_key,
	)


	class DauphinScalarUnionConfigType(ConfigTypeMixin, dauphin.ObjectType):			class GrapheneScalarUnionConfigType(ConfigTypeMixin, graphene.ObjectType):
				scalar_type = graphene.NonNull(GrapheneConfigType)
				non_scalar_type = graphene.NonNull(GrapheneConfigType)
				scalar_type_key = graphene.NonNull(graphene.String)
				non_scalar_type_key = graphene.NonNull(graphene.String)

	class Meta:			class Meta:
				interfaces = (GrapheneConfigType,)
	name = "ScalarUnionConfigType"			name = "ScalarUnionConfigType"
	interfaces = [DauphinConfigType]

	scalar_type = dauphin.NonNull(DauphinConfigType)
	non_scalar_type = dauphin.NonNull(DauphinConfigType)

	scalar_type_key = dauphin.NonNull(dauphin.String)
	non_scalar_type_key = dauphin.NonNull(dauphin.String)

	def get_scalar_type_key(self):			def get_scalar_type_key(self):
	return self._config_type_snap.scalar_type_key			return self._config_type_snap.scalar_type_key

	def get_non_scalar_type_key(self):			def get_non_scalar_type_key(self):
	return self._config_type_snap.non_scalar_type_key			return self._config_type_snap.non_scalar_type_key

	def resolve_scalar_type_key(self, _):			def resolve_scalar_type_key(self, _):
	return self.get_scalar_type_key()			return self.get_scalar_type_key()

	def resolve_non_scalar_type_key(self, _):			def resolve_non_scalar_type_key(self, _):
	return self.get_non_scalar_type_key()			return self.get_non_scalar_type_key()

	def resolve_scalar_type(self, _):			def resolve_scalar_type(self, _):
	return to_dauphin_config_type(self._config_schema_snapshot, self.get_scalar_type_key())			return to_config_type(self._config_schema_snapshot, self.get_scalar_type_key())

	def resolve_non_scalar_type(self, _):			def resolve_non_scalar_type(self, _):
	return to_dauphin_config_type(self._config_schema_snapshot, self.get_non_scalar_type_key())			return to_config_type(self._config_schema_snapshot, self.get_non_scalar_type_key())


	class DauphinNullableConfigType(ConfigTypeMixin, dauphin.ObjectType):			class GrapheneNullableConfigType(ConfigTypeMixin, graphene.ObjectType):
	class Meta:			class Meta:
				interfaces = (GrapheneConfigType, GrapheneWrappingConfigType)
	name = "NullableConfigType"			name = "NullableConfigType"
	interfaces = [DauphinConfigType, DauphinWrappingConfigType]

	def resolve_of_type(self, _graphene_info):			def resolve_of_type(self, _graphene_info):
	return to_dauphin_config_type(			return to_config_type(self._config_schema_snapshot, self._config_type_snap.inner_type_key)
	self._config_schema_snapshot, self._config_type_snap.inner_type_key
	)
				class GrapheneEnumConfigValue(graphene.ObjectType):
				value = graphene.NonNull(graphene.String)
				description = graphene.String()

				class Meta:
				name = "EnumConfigValue"


	class DauphinEnumConfigType(ConfigTypeMixin, dauphin.ObjectType):			class GrapheneEnumConfigType(ConfigTypeMixin, graphene.ObjectType):
	class Meta:			class Meta:
				interfaces = (GrapheneConfigType,)
	name = "EnumConfigType"			name = "EnumConfigType"
	interfaces = [DauphinConfigType]

	values = dauphin.non_null_list("EnumConfigValue")			values = non_null_list(GrapheneEnumConfigValue)
	given_name = dauphin.NonNull(dauphin.String)			given_name = graphene.NonNull(graphene.String)

	def resolve_values(self, _graphene_info):			def resolve_values(self, _graphene_info):
	return [			return [
	DauphinEnumConfigValue(value=ev.value, description=ev.description)			GrapheneEnumConfigValue(value=ev.value, description=ev.description)
	for ev in self._config_type_snap.enum_values			for ev in self._config_type_snap.enum_values
	]			]

	def resolve_given_name(self, _):			def resolve_given_name(self, _):
	return self._config_type_snap.given_name			return self._config_type_snap.given_name


	class DauphinEnumConfigValue(dauphin.ObjectType):			class GrapheneConfigTypeField(graphene.ObjectType):
	class Meta:			name = graphene.NonNull(graphene.String)
	name = "EnumConfigValue"			description = graphene.String()
				config_type = graphene.NonNull(GrapheneConfigType)
				config_type_key = graphene.NonNull(graphene.String)
				is_required = graphene.NonNull(graphene.Boolean)

	value = dauphin.NonNull(dauphin.String)
	description = dauphin.String()


	class DauphinCompositeConfigType(ConfigTypeMixin, dauphin.ObjectType):
	class Meta:
	name = "CompositeConfigType"
	interfaces = [DauphinConfigType]

	fields = dauphin.non_null_list("ConfigTypeField")

	def resolve_fields(self, _graphene_info):
	return sorted(
	[
	DauphinConfigTypeField(
	config_schema_snapshot=self._config_schema_snapshot, field_snap=field_snap,
	)
	for field_snap in self._config_type_snap.fields
	],
	key=lambda field: field.name,
	)


	class DauphinConfigTypeField(dauphin.ObjectType):
	class Meta:			class Meta:
	name = "ConfigTypeField"			name = "ConfigTypeField"

	name = dauphin.NonNull(dauphin.String)
	description = dauphin.String()
	config_type = dauphin.NonNull("ConfigType")
	config_type_key = dauphin.NonNull(dauphin.String)
	is_required = dauphin.NonNull(dauphin.Boolean)

	def resolve_config_type_key(self, _):			def resolve_config_type_key(self, _):
	return self._field_snap.type_key			return self._field_snap.type_key

	def __init__(self, config_schema_snapshot, field_snap):			def __init__(self, config_schema_snapshot, field_snap):
	self._config_schema_snapshot = check.inst_param(			self._config_schema_snapshot = check.inst_param(
	config_schema_snapshot, "config_schema_snapshot", ConfigSchemaSnapshot			config_schema_snapshot, "config_schema_snapshot", ConfigSchemaSnapshot
	)			)
	self._field_snap = check.inst_param(field_snap, "field_snap", ConfigFieldSnap)			self._field_snap = check.inst_param(field_snap, "field_snap", ConfigFieldSnap)
	super(DauphinConfigTypeField, self).__init__(			super().__init__(
	name=field_snap.name,			name=field_snap.name,
	description=field_snap.description,			description=field_snap.description,
	is_required=field_snap.is_required,			is_required=field_snap.is_required,
	)			)

	def resolve_config_type(self, _graphene_info):			def resolve_config_type(self, _graphene_info):
	return to_dauphin_config_type(self._config_schema_snapshot, self._field_snap.type_key)			return to_config_type(self._config_schema_snapshot, self._field_snap.type_key)


				class GrapheneCompositeConfigType(ConfigTypeMixin, graphene.ObjectType):
				fields = non_null_list(GrapheneConfigTypeField)

				class Meta:
				interfaces = (GrapheneConfigType,)
				name = "CompositeConfigType"

				def resolve_fields(self, _graphene_info):
				return sorted(
				[
				GrapheneConfigTypeField(
				config_schema_snapshot=self._config_schema_snapshot, field_snap=field_snap,
				)
				for field_snap in self._config_type_snap.fields
				],
				key=lambda field: field.name,
				)


				types = [
				GrapheneArrayConfigType,
				GrapheneCompositeConfigType,
				GrapheneConfigType,
				GrapheneConfigTypeField,
				GrapheneEnumConfigType,
				GrapheneEnumConfigValue,
				GrapheneNullableConfigType,
				GrapheneRegularConfigType,
				GrapheneScalarUnionConfigType,
				GrapheneWrappingConfigType,
				]
				maxAuthorUnsubmitted Done Inline Actions could also be `__types__` or `__all__` max: could also be `__types__` or `__all__`