Add ConfigSchema type alias (#7596)

python_modules/dagster/dagster/config/config_schema.py

@@ -1,3 +1,28 @@
+from typing import TYPE_CHECKING, Dict, List, Type, Union
+
+from typing_extensions import TypeAlias
+
+if TYPE_CHECKING:
+    from dagster.config.config_type import ConfigType
+    from dagster.config.field import Field
+
+# Eventually, the below `ConfigSchemaType` should be renamed to `ConfigSchema` and the class
+# definition should be dropped. The reason we don't do this now is that sphinx autodoc doesn't
+# support type aliases, so there is no good way to gracefully attach a docstring to this and have it
+# show up in the docs. See: https://github.com/sphinx-doc/sphinx/issues/8934
+#
+# Unfortunately mypy doesn't support recursive types, which would be used to properly define the
+# List/Dict elements of this union: `Dict[str, ConfigSchema]`, `List[ConfigSchema]`.
+ConfigSchemaType: TypeAlias = Union[
+    Type[Union[bool, float, int, str]],
+    Type[Union[dict, list]],
+    "ConfigType",
+    "Field",
+    Dict[str, object],
+    List[object],
+]
+
+
 class ConfigSchema:
     """This is a placeholder type. Any time that it appears in documentation, it means that any of
     the following types are acceptable:

python_modules/dagster/dagster/config/config_type.py

@@ -4,6 +4,7 @@
 
 from dagster import check
 from dagster.builtins import BuiltinEnum
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.serdes import whitelist_for_serdes
 
 
@@ -367,7 +368,10 @@ class ScalarUnion(ConfigType):
     """
 
     def __init__(
-        self, scalar_type: typing.Any, non_scalar_schema: typing.Any, _key: Optional[str] = None
+        self,
+        scalar_type: typing.Any,
+        non_scalar_schema: ConfigSchemaType,
+        _key: Optional[str] = None,
     ):
         from .field import resolve_to_config_type
 

python_modules/dagster/dagster/config/field.py

@@ -2,6 +2,7 @@
 
 from dagster import check
 from dagster.builtins import BuiltinEnum
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.core.errors import DagsterInvalidConfigError, DagsterInvalidDefinitionError
 from dagster.serdes import serialize_value
 from dagster.utils import is_enum_value
@@ -35,7 +36,7 @@ def helpful_list_error_string():
 
 
 @overload
-def resolve_to_config_type(dagster_type: ConfigType) -> ConfigType:
+def resolve_to_config_type(dagster_type: Union[ConfigType, ConfigSchemaType]) -> ConfigType:
     pass
 
 
@@ -44,7 +45,7 @@ def resolve_to_config_type(dagster_type: object) -> Union[ConfigType, bool]:
     pass
 
 
-def resolve_to_config_type(dagster_type) -> Union[ConfigType, bool]:
+def resolve_to_config_type(dagster_type: object) -> Union[ConfigType, bool]:
     from .field_utils import convert_fields_to_dict_type
 
     # Short circuit if it's already a Config Type

python_modules/dagster/dagster/core/definitions/decorators/composite_solid_decorator.py

@@ -1,7 +1,8 @@
 from functools import update_wrapper
-from typing import Any, Callable, Dict, List, Optional, Union, overload
+from typing import Any, Callable, List, Optional, Union, overload
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.core.decorator_utils import format_docstring_for_description
 
 from ..composition import do_composition, get_validated_config_mapping
@@ -17,7 +18,7 @@ def __init__(
         input_defs: Optional[List[InputDefinition]] = None,
         output_defs: Optional[List[OutputDefinition]] = None,
         description: Optional[str] = None,
-        config_schema: Optional[Any] = None,
+        config_schema: Optional[ConfigSchemaType] = None,
         config_fn: Optional[Callable[[dict], dict]] = None,
     ):
         self.name = check.opt_str_param(name, "name")
@@ -82,7 +83,7 @@ def composite_solid(
     input_defs: Optional[List[InputDefinition]] = ...,
     output_defs: Optional[List[OutputDefinition]] = ...,
     description: Optional[str] = ...,
-    config_schema: Optional[Dict[str, Any]] = ...,
+    config_schema: Optional[ConfigSchemaType] = ...,
     config_fn: Optional[Callable[[dict], dict]] = ...,
 ) -> _CompositeSolid:
     ...
@@ -93,7 +94,7 @@ def composite_solid(
     input_defs: Optional[List[InputDefinition]] = None,
     output_defs: Optional[List[OutputDefinition]] = None,
     description: Optional[str] = None,
-    config_schema: Optional[Dict[str, Any]] = None,
+    config_schema: Optional[ConfigSchemaType] = None,
     config_fn: Optional[Callable[[dict], dict]] = None,
 ) -> Union[CompositeSolidDefinition, _CompositeSolid]:
     """Create a composite solid with the specified parameters from the decorated composition

python_modules/dagster/dagster/core/definitions/decorators/config_mapping_decorator.py

@@ -1,14 +1,15 @@
 from typing import Any, Callable, Optional, Union, overload
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 
 from ..config import ConfigMapping
 
 
 class _ConfigMapping:
     def __init__(
         self,
-        config_schema: Optional[Any] = None,
+        config_schema: Optional[ConfigSchemaType] = None,
         receive_processed_config_values: Optional[bool] = None,
     ):
         self.config_schema = config_schema
@@ -36,15 +37,15 @@ def config_mapping(
 @overload
 def config_mapping(
     config_fn: None = ...,
-    config_schema: Any = ...,
+    config_schema: ConfigSchemaType = ...,
     receive_processed_config_values: Optional[bool] = ...,
 ) -> Union[_ConfigMapping, ConfigMapping]:
     ...
 
 
 def config_mapping(
     config_fn: Optional[Callable[..., Any]] = None,
-    config_schema: Any = None,
+    config_schema: Optional[ConfigSchemaType] = None,
     receive_processed_config_values: Optional[bool] = None,
 ) -> Union[ConfigMapping, _ConfigMapping]:
     """Create a config mapping with the specified parameters from the decorated function.

python_modules/dagster/dagster/core/definitions/decorators/op_decorator.py

@@ -13,6 +13,7 @@
 )
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.core.decorator_utils import format_docstring_for_description
 
 from ....seven.typing import get_origin
@@ -187,7 +188,7 @@ def op(
     description: Optional[str] = ...,
     ins: Optional[Dict[str, In]] = ...,
     out: Optional[Union[Out, Dict[str, Out]]] = ...,
-    config_schema: Optional[Union[Any, Dict[str, Any]]] = ...,
+    config_schema: Optional[ConfigSchemaType] = ...,
     required_resource_keys: Optional[Set[str]] = ...,
     tags: Optional[Dict[str, Any]] = ...,
     version: Optional[str] = ...,
@@ -203,7 +204,7 @@ def op(
     description: Optional[str] = None,
     ins: Optional[Dict[str, In]] = None,
     out: Optional[Union[Out, Dict[str, Out]]] = None,
-    config_schema: Optional[Union[Any, Dict[str, Any]]] = None,
+    config_schema: Optional[ConfigSchemaType] = None,
     required_resource_keys: Optional[Set[str]] = None,
     tags: Optional[Dict[str, Any]] = None,
     version: Optional[str] = None,

python_modules/dagster/dagster/core/definitions/decorators/pipeline_decorator.py

@@ -2,6 +2,7 @@
 from typing import Any, Callable, Dict, List, Optional, Set, Union, overload
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.core.decorator_utils import format_docstring_for_description
 from dagster.core.definitions.policy import RetryPolicy
 from dagster.utils.backcompat import experimental_arg_warning
@@ -27,7 +28,7 @@ def __init__(
         hook_defs: Optional[Set[HookDefinition]] = None,
         input_defs: Optional[List[InputDefinition]] = None,
         output_defs: Optional[List[OutputDefinition]] = None,
-        config_schema: Optional[Dict[str, Any]] = None,
+        config_schema: Optional[ConfigSchemaType] = None,
         config_fn: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None,
         solid_retry_policy: Optional[RetryPolicy] = None,
         version_strategy: Optional[VersionStrategy] = None,
@@ -124,7 +125,7 @@ def pipeline(
     hook_defs: Optional[Set[HookDefinition]] = ...,
     input_defs: Optional[List[InputDefinition]] = ...,
     output_defs: Optional[List[OutputDefinition]] = ...,
-    config_schema: Optional[Dict[str, Any]] = ...,
+    config_schema: Optional[ConfigSchemaType] = ...,
     config_fn: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = ...,
     solid_retry_policy: Optional[RetryPolicy] = ...,
     version_strategy: Optional[VersionStrategy] = ...,
@@ -141,7 +142,7 @@ def pipeline(
     hook_defs: Optional[Set[HookDefinition]] = None,
     input_defs: Optional[List[InputDefinition]] = None,
     output_defs: Optional[List[OutputDefinition]] = None,
-    config_schema: Optional[Dict[str, Any]] = None,
+    config_schema: Optional[ConfigSchemaType] = None,
     config_fn: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None,
     solid_retry_policy: Optional[RetryPolicy] = None,
     version_strategy: Optional[VersionStrategy] = None,

python_modules/dagster/dagster/core/definitions/decorators/solid_decorator.py

@@ -14,6 +14,7 @@
 )
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.core.decorator_utils import format_docstring_for_description
 from dagster.core.errors import DagsterInvalidDefinitionError
 from dagster.core.types.dagster_type import DagsterTypeKind
@@ -74,7 +75,7 @@ def __init__(
         output_defs: Optional[Sequence[OutputDefinition]] = None,
         description: Optional[str] = None,
         required_resource_keys: Optional[Set[str]] = None,
-        config_schema: Optional[Union[Any, Dict[str, Any]]] = None,
+        config_schema: Optional[ConfigSchemaType] = None,
         tags: Optional[Dict[str, Any]] = None,
         version: Optional[str] = None,
         decorator_takes_context: Optional[bool] = True,
@@ -155,7 +156,7 @@ def solid(
     description: Optional[str] = ...,
     input_defs: Optional[Sequence[InputDefinition]] = ...,
     output_defs: Optional[Sequence[OutputDefinition]] = ...,
-    config_schema: Optional[Union[Any, Dict[str, Any]]] = ...,
+    config_schema: Optional[ConfigSchemaType] = ...,
     required_resource_keys: Optional[Set[str]] = ...,
     tags: Optional[Dict[str, Any]] = ...,
     version: Optional[str] = ...,
@@ -169,7 +170,7 @@ def solid(
     description: Optional[str] = None,
     input_defs: Optional[Sequence[InputDefinition]] = None,
     output_defs: Optional[Sequence[OutputDefinition]] = None,
-    config_schema: Optional[Union[Any, Dict[str, Any]]] = None,
+    config_schema: Optional[ConfigSchemaType] = None,
     required_resource_keys: Optional[Set[str]] = None,
     tags: Optional[Dict[str, Any]] = None,
     version: Optional[str] = None,

python_modules/dagster/dagster/core/definitions/definition_config_schema.py

@@ -2,6 +2,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Union
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.config.config_type import ConfigAnyInstance, ConfigType
 from dagster.config.evaluate_value_result import EvaluateValueResult
 from dagster.config.field import Field
@@ -14,7 +15,7 @@
 
 
 def convert_user_facing_definition_config_schema(
-    potential_schema: Union["IDefinitionConfigSchema", Dict[str, Any], None],
+    potential_schema: Optional[Union["IDefinitionConfigSchema", ConfigSchemaType]]
 ) -> "IDefinitionConfigSchema":
     if potential_schema is None:
         return DefinitionConfigSchema(Field(ConfigAnyInstance, is_required=False))

python_modules/dagster/dagster/core/definitions/solid_definition.py

@@ -15,6 +15,7 @@
 )
 
 from dagster import check
+from dagster.config.config_schema import ConfigSchemaType
 from dagster.core.definitions.dependency import NodeHandle
 from dagster.core.definitions.policy import RetryPolicy
 from dagster.core.errors import DagsterInvalidDefinitionError, DagsterInvalidInvocationError
@@ -98,7 +99,7 @@ def __init__(
         input_defs: Sequence[InputDefinition],
         compute_fn: Union[Callable[..., Any], "DecoratedSolidFunction"],
         output_defs: Sequence[OutputDefinition],
-        config_schema: Optional[Union[Dict[str, Any], IDefinitionConfigSchema]] = None,
+        config_schema: Optional[Union[ConfigSchemaType, IDefinitionConfigSchema]] = None,
         description: Optional[str] = None,
         tags: Optional[Dict[str, str]] = None,
         required_resource_keys: Optional[Union[Set[str], FrozenSet[str]]] = None,