Differential D6164 Diff 30381 python_modules/dagster-graphql/dagster_graphql/schema/dagster_types.py

Changeset View

Standalone View

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

				import graphene
	from dagster import check			from dagster import check
	from dagster.core.snap import PipelineSnapshot			from dagster.core.snap import PipelineSnapshot
	from dagster.core.types.dagster_type import DagsterTypeKind			from dagster.core.types.dagster_type import DagsterTypeKind
	from dagster_graphql import dauphin

	from .config_types import DauphinConfigType, to_dauphin_config_type			from .config_types import ConfigType, to_config_type
				from .errors import DagsterTypeNotFoundError, PipelineNotFoundError, PythonError
				from .util import non_null_list


	def config_type_for_schema(pipeline_snapshot, schema_key):			def config_type_for_schema(pipeline_snapshot, schema_key):
	return (			return (
	to_dauphin_config_type(pipeline_snapshot.config_schema_snapshot, schema_key)			to_config_type(pipeline_snapshot.config_schema_snapshot, schema_key) if schema_key else None
	if schema_key
	else None
	)			)


	def to_dauphin_dagster_type(pipeline_snapshot, dagster_type_key):			def to_dagster_type(pipeline_snapshot, dagster_type_key):
	check.str_param(dagster_type_key, "dagster_type_key")			check.str_param(dagster_type_key, "dagster_type_key")
	check.inst_param(pipeline_snapshot, pipeline_snapshot, PipelineSnapshot)			check.inst_param(pipeline_snapshot, pipeline_snapshot, PipelineSnapshot)

	dagster_type_meta = pipeline_snapshot.dagster_type_namespace_snapshot.get_dagster_type_snap(			dagster_type_meta = pipeline_snapshot.dagster_type_namespace_snapshot.get_dagster_type_snap(
	dagster_type_key			dagster_type_key
	)			)

	base_args = dict(			base_args = dict(
	key=dagster_type_meta.key,			key=dagster_type_meta.key,
	name=dagster_type_meta.name,			name=dagster_type_meta.name,
	display_name=dagster_type_meta.display_name,			display_name=dagster_type_meta.display_name,
	description=dagster_type_meta.description,			description=dagster_type_meta.description,
	is_builtin=dagster_type_meta.is_builtin,			is_builtin=dagster_type_meta.is_builtin,
	is_nullable=dagster_type_meta.kind == DagsterTypeKind.NULLABLE,			is_nullable=dagster_type_meta.kind == DagsterTypeKind.NULLABLE,
	is_list=dagster_type_meta.kind == DagsterTypeKind.LIST,			is_list=dagster_type_meta.kind == DagsterTypeKind.LIST,
	is_nothing=dagster_type_meta.kind == DagsterTypeKind.NOTHING,			is_nothing=dagster_type_meta.kind == DagsterTypeKind.NOTHING,
	input_schema_type=config_type_for_schema(			input_schema_type=config_type_for_schema(
	pipeline_snapshot, dagster_type_meta.loader_schema_key,			pipeline_snapshot, dagster_type_meta.loader_schema_key,
	),			),
	output_schema_type=config_type_for_schema(			output_schema_type=config_type_for_schema(
	pipeline_snapshot, dagster_type_meta.materializer_schema_key,			pipeline_snapshot, dagster_type_meta.materializer_schema_key,
	),			),
	inner_types=list(			inner_types=list(
	map(			map(
	lambda key: to_dauphin_dagster_type(pipeline_snapshot, key),			lambda key: to_dagster_type(pipeline_snapshot, key),
	dagster_type_meta.type_param_keys,			dagster_type_meta.type_param_keys,
	)			)
	),			),
	)			)

	if dagster_type_meta.kind == DagsterTypeKind.LIST:			if dagster_type_meta.kind == DagsterTypeKind.LIST:
	base_args["of_type"] = to_dauphin_dagster_type(			base_args["of_type"] = to_dagster_type(
	pipeline_snapshot, dagster_type_meta.type_param_keys[0]			pipeline_snapshot, dagster_type_meta.type_param_keys[0]
	)			)
	return DauphinListDagsterType(**base_args)			return ListDagsterType(**base_args)
	elif dagster_type_meta.kind == DagsterTypeKind.NULLABLE:			elif dagster_type_meta.kind == DagsterTypeKind.NULLABLE:
	base_args["of_type"] = to_dauphin_dagster_type(			base_args["of_type"] = to_dagster_type(
	pipeline_snapshot, dagster_type_meta.type_param_keys[0]			pipeline_snapshot, dagster_type_meta.type_param_keys[0]
	)			)
	return DauphinNullableDagsterType(**base_args)			return NullableDagsterType(**base_args)
	else:			else:
	return DauphinRegularDagsterType(**base_args)			return RegularDagsterType(**base_args)


	class DauphinDagsterType(dauphin.Interface):			class DagsterType(graphene.Interface):
				# This is necessary because the type has a field called `name`
	class Meta:			class Meta:
	name = "DagsterType"			name = "DagsterType"

	key = dauphin.NonNull(dauphin.String)			key = graphene.NonNull(graphene.String)
	name = dauphin.String()			name = graphene.String()
	display_name = dauphin.NonNull(dauphin.String)			display_name = graphene.NonNull(graphene.String)
	description = dauphin.String()			description = graphene.String()

	is_nullable = dauphin.NonNull(dauphin.Boolean)			is_nullable = graphene.NonNull(graphene.Boolean)
	is_list = dauphin.NonNull(dauphin.Boolean)			is_list = graphene.NonNull(graphene.Boolean)
	is_builtin = dauphin.NonNull(dauphin.Boolean)			is_builtin = graphene.NonNull(graphene.Boolean)
	is_nothing = dauphin.NonNull(dauphin.Boolean)			is_nothing = graphene.NonNull(graphene.Boolean)

	input_schema_type = dauphin.Field(DauphinConfigType)			input_schema_type = graphene.Field(ConfigType)
	output_schema_type = dauphin.Field(DauphinConfigType)			output_schema_type = graphene.Field(ConfigType)

	inner_types = dauphin.non_null_list("DagsterType")			inner_types = non_null_list(lambda: DagsterType)


	class DauphinRegularDagsterType(dauphin.ObjectType):			class RegularDagsterType(graphene.ObjectType):
	class Meta:			class Meta:
	name = "RegularDagsterType"			interfaces = (DagsterType,)
	interfaces = [DauphinDagsterType]


	class DauphinWrappingDagsterType(dauphin.Interface):			class WrappingDagsterType(graphene.Interface):
	class Meta:			of_type = graphene.Field(graphene.NonNull(DagsterType))
	name = "WrappingDagsterType"

	of_type = dauphin.Field(dauphin.NonNull(DauphinDagsterType))			class ListDagsterType(graphene.ObjectType):
				class Meta:
				interfaces = (DagsterType, WrappingDagsterType)


	class DauphinListDagsterType(dauphin.ObjectType):			class NullableDagsterType(graphene.ObjectType):
	class Meta:			class Meta:
	name = "ListDagsterType"			interfaces = (DagsterType, WrappingDagsterType)
	interfaces = [DauphinDagsterType, DauphinWrappingDagsterType]


	class DauphinNullableDagsterType(dauphin.ObjectType):			class DagsterTypeOrError(graphene.Union):
	class Meta:			class Meta:
	name = "NullableDagsterType"			types = (
	interfaces = [DauphinDagsterType, DauphinWrappingDagsterType]			RegularDagsterType,
				PipelineNotFoundError,
				DagsterTypeNotFoundError,
				PythonError,
				)


				types = [
				DagsterType,
				DagsterTypeOrError,
				ListDagsterType,
				NullableDagsterType,
				RegularDagsterType,
				WrappingDagsterType,
				]