Differential D6164 Diff 30546 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 GrapheneConfigType, to_config_type
				from .errors import (
				GrapheneDagsterTypeNotFoundError,
				GraphenePipelineNotFoundError,
				GraphenePythonError,
				)
				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 GrapheneListDagsterType(**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 GrapheneNullableDagsterType(**base_args)
	else:			else:
	return DauphinRegularDagsterType(**base_args)			return GrapheneRegularDagsterType(**base_args)


	class DauphinDagsterType(dauphin.Interface):			class GrapheneDagsterType(graphene.Interface):
	class Meta:			key = graphene.NonNull(graphene.String)
	name = "DagsterType"			name = graphene.String()
				display_name = graphene.NonNull(graphene.String)
				description = graphene.String()

	key = dauphin.NonNull(dauphin.String)			is_nullable = graphene.NonNull(graphene.Boolean)
	name = dauphin.String()			is_list = graphene.NonNull(graphene.Boolean)
	display_name = dauphin.NonNull(dauphin.String)			is_builtin = graphene.NonNull(graphene.Boolean)
	description = dauphin.String()			is_nothing = graphene.NonNull(graphene.Boolean)

	is_nullable = dauphin.NonNull(dauphin.Boolean)			input_schema_type = graphene.Field(GrapheneConfigType)
	is_list = dauphin.NonNull(dauphin.Boolean)			output_schema_type = graphene.Field(GrapheneConfigType)
	is_builtin = dauphin.NonNull(dauphin.Boolean)
	is_nothing = dauphin.NonNull(dauphin.Boolean)

	input_schema_type = dauphin.Field(DauphinConfigType)			inner_types = non_null_list(lambda: GrapheneDagsterType)
	output_schema_type = dauphin.Field(DauphinConfigType)

	inner_types = dauphin.non_null_list("DagsterType")			class Meta:
				name = "DagsterType"


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


	class DauphinWrappingDagsterType(dauphin.Interface):			class GrapheneWrappingDagsterType(graphene.Interface):
				of_type = graphene.Field(graphene.NonNull(GrapheneDagsterType))

	class Meta:			class Meta:
	name = "WrappingDagsterType"			name = "WrappingDagsterType"

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


	class DauphinListDagsterType(dauphin.ObjectType):			class GrapheneListDagsterType(graphene.ObjectType):
	class Meta:			class Meta:
				interfaces = (GrapheneDagsterType, GrapheneWrappingDagsterType)
	name = "ListDagsterType"			name = "ListDagsterType"
	interfaces = [DauphinDagsterType, DauphinWrappingDagsterType]


	class DauphinNullableDagsterType(dauphin.ObjectType):			class GrapheneNullableDagsterType(graphene.ObjectType):
	class Meta:			class Meta:
				interfaces = (GrapheneDagsterType, GrapheneWrappingDagsterType)
	name = "NullableDagsterType"			name = "NullableDagsterType"
	interfaces = [DauphinDagsterType, DauphinWrappingDagsterType]

				class GrapheneDagsterTypeOrError(graphene.Union):
				class Meta:
				types = (
				GrapheneRegularDagsterType,
				GraphenePipelineNotFoundError,
				GrapheneDagsterTypeNotFoundError,
				GraphenePythonError,
				)
				name = "DagsterTypeOrError"


				types = [
				GrapheneDagsterType,
				GrapheneDagsterTypeOrError,
				GrapheneListDagsterType,
				GrapheneNullableDagsterType,
				GrapheneRegularDagsterType,
				GrapheneWrappingDagsterType,
				]