make all optional params explicit (#6934)

python_modules/dagster/dagster/check/__init__.py

@@ -111,7 +111,7 @@ def opt_callable_param(
     return default if obj is None else obj
 
 
-def is_callable(obj: object, desc: str = None) -> Callable:
+def is_callable(obj: object, desc: Optional[str] = None) -> Callable:
     if not callable(obj):
         raise CheckError(f"Must be callable. Got {obj}.{desc and f' Description: {desc}.' or ''}")
     return obj
@@ -317,9 +317,9 @@ def opt_dict_elem(ddict: Dict, key: str) -> Dict:
 
 def is_dict(
     obj: object,
-    key_type: TypeOrTupleOfTypes = None,
-    value_type: TypeOrTupleOfTypes = None,
-    desc: str = None,
+    key_type: Optional[TypeOrTupleOfTypes] = None,
+    value_type: Optional[TypeOrTupleOfTypes] = None,
+    desc: Optional[str] = None,
 ) -> Dict:
     from dagster.utils import frozendict
 
@@ -334,8 +334,8 @@ def is_dict(
 
 def _check_key_value_types(
     obj_dict: Dict,
-    key_type: TypeOrTupleOfTypes = None,
-    value_type: TypeOrTupleOfTypes = None,
+    key_type: Optional[TypeOrTupleOfTypes] = None,
+    value_type: Optional[TypeOrTupleOfTypes] = None,
     key_check: Callable = isinstance,
     value_check: Callable = isinstance,
 ) -> Dict:
@@ -362,9 +362,9 @@ def _check_key_value_types(
 
 def _check_two_dim_key_value_types(
     obj_dict: Dict,
-    key_type: TypeOrTupleOfTypes = None,
-    _param_name: str = None,
-    value_type: TypeOrTupleOfTypes = None,
+    key_type: Optional[TypeOrTupleOfTypes] = None,
+    _param_name: Optional[str] = None,
+    value_type: Optional[TypeOrTupleOfTypes] = None,
 ) -> Dict:
     _check_key_value_types(obj_dict, key_type, dict)  # check level one
 
@@ -532,7 +532,7 @@ def opt_int_elem(ddict: Dict, key: str) -> Optional[int]:
 
 
 def inst_param(
-    obj: T, param_name: str, ttype: TypeOrTupleOfTypes, additional_message: str = None
+    obj: T, param_name: str, ttype: TypeOrTupleOfTypes, additional_message: Optional[str] = None
 ) -> T:
     if not isinstance(obj, ttype):
         raise _param_type_mismatch_exception(
@@ -552,21 +552,21 @@ def opt_inst_param(obj: T, param_name: str, ttype: TypeOrTupleOfTypes) -> T:
 
 
 def opt_inst_param(
-    obj: T, param_name: str, ttype: TypeOrTupleOfTypes, default: T = None
+    obj: T, param_name: str, ttype: TypeOrTupleOfTypes, default: Optional[T] = None
 ) -> Optional[T]:
     if obj is not None and not isinstance(obj, ttype):
         raise _param_type_mismatch_exception(obj, ttype, param_name)
     return default if obj is None else obj
 
 
-def inst(obj: T, ttype: TypeOrTupleOfTypes, desc: str = None) -> T:
+def inst(obj: T, ttype: TypeOrTupleOfTypes, desc: Optional[str] = None) -> T:
     if not isinstance(obj, ttype):
         raise _type_mismatch_error(obj, ttype, desc)
     return obj
 
 
 def opt_inst(
-    obj: T, ttype: TypeOrTupleOfTypes, desc: str = None, default: Optional[T] = None
+    obj: T, ttype: TypeOrTupleOfTypes, desc: Optional[str] = None, default: Optional[T] = None
 ) -> Optional[T]:
     if obj is not None and not isinstance(obj, ttype):
         raise _type_mismatch_error(obj, ttype, desc)
@@ -578,7 +578,7 @@ def opt_inst(
 # ########################
 
 
-def list_param(obj: object, param_name: str, of_type: TypeOrTupleOfTypes = None) -> List:
+def list_param(obj: object, param_name: str, of_type: Optional[TypeOrTupleOfTypes] = None) -> List:
     from dagster.utils import frozenlist
 
     if not isinstance(obj, (frozenlist, list)):
@@ -590,7 +590,9 @@ def list_param(obj: object, param_name: str, of_type: TypeOrTupleOfTypes = None)
     return _check_list_items(obj, of_type)
 
 
-def opt_list_param(obj: object, param_name: str, of_type: TypeOrTupleOfTypes = None) -> List:
+def opt_list_param(
+    obj: object, param_name: str, of_type: Optional[TypeOrTupleOfTypes] = None
+) -> List:
     """Ensures argument obj is a list or None; in the latter case, instantiates an empty list
     and returns it.
 
@@ -738,7 +740,7 @@ def not_none_param(obj: Optional[T], param_name: str) -> T:
     return obj
 
 
-def not_none(value: Optional[T], desc: str = None) -> T:
+def not_none(value: Optional[T], desc: Optional[str] = None) -> T:
     if value is None:
         raise ValueError(f"Expected non-None value: {desc}")
     return value
@@ -778,7 +780,9 @@ def opt_numeric_param(
 # ########################
 
 
-def set_param(obj: object, param_name: str, of_type: TypeOrTupleOfTypes = None) -> AbstractSet:
+def set_param(
+    obj: object, param_name: str, of_type: Optional[TypeOrTupleOfTypes] = None
+) -> AbstractSet:
     if not isinstance(obj, (frozenset, set)):
         raise _param_type_mismatch_exception(obj, (frozenset, set), param_name)
 
@@ -788,7 +792,9 @@ def set_param(obj: object, param_name: str, of_type: TypeOrTupleOfTypes = None)
     return _check_set_items(obj, of_type)
 
 
-def opt_set_param(obj: object, param_name: str, of_type: TypeOrTupleOfTypes = None) -> AbstractSet:
+def opt_set_param(
+    obj: object, param_name: str, of_type: Optional[TypeOrTupleOfTypes] = None
+) -> AbstractSet:
     """Ensures argument obj is a set or None; in the latter case, instantiates an empty set
     and returns it.
 
@@ -1049,12 +1055,12 @@ def _check_tuple_items(
 # ###################################################################################################
 
 
-def param_invariant(condition: Any, param_name: str, desc: str = None):
+def param_invariant(condition: Any, param_name: str, desc: Optional[str] = None):
     if not condition:
         raise _param_invariant_exception(param_name, desc)
 
 
-def invariant(condition: Any, desc: str = None) -> bool:
+def invariant(condition: Any, desc: Optional[str] = None) -> bool:
     if not condition:
         if desc:
             raise CheckError(f"Invariant failed. Description: {desc}")

python_modules/dagster/dagster/config/config_type.py

@@ -223,7 +223,10 @@ class EnumValue:
     """
 
     def __init__(
-        self, config_value: str, python_value: object = None, description: Optional[str] = None
+        self,
+        config_value: str,
+        python_value: Optional[object] = None,
+        description: Optional[str] = None,
     ):
         self.config_value = check.str_param(config_value, "config_value")
         self.python_value = config_value if python_value is None else python_value

python_modules/dagster/dagster/core/asset_defs/assets_job.py

@@ -38,7 +38,7 @@ def build_assets_job(
     source_assets: Optional[Sequence[Union[SourceAsset, AssetsDefinition]]] = None,
     resource_defs: Optional[Dict[str, ResourceDefinition]] = None,
     description: Optional[str] = None,
-    config: Union[ConfigMapping, Dict[str, Any], PartitionedConfig] = None,
+    config: Optional[Union[ConfigMapping, Dict[str, Any], PartitionedConfig]] = None,
     tags: Optional[Dict[str, Any]] = None,
     executor_def: Optional[ExecutorDefinition] = None,
 ) -> JobDefinition:

python_modules/dagster/dagster/core/asset_defs/decorators.py

@@ -59,7 +59,7 @@ def asset(
 
 @experimental_decorator
 def asset(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], Optional[str]]] = None,
     namespace: Optional[Sequence[str]] = None,
     ins: Optional[Mapping[str, AssetIn]] = None,
     non_argument_deps: Optional[Set[AssetKey]] = None,

python_modules/dagster/dagster/core/definitions/composition.py

@@ -564,7 +564,7 @@ def to_job(
         name: Optional[str] = None,
         description: Optional[str] = None,
         resource_defs: Optional[Dict[str, ResourceDefinition]] = None,
-        config: Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"] = None,
+        config: Optional[Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"]] = None,
         tags: Optional[Dict[str, Any]] = None,
         logger_defs: Optional[Dict[str, LoggerDefinition]] = None,
         executor_def: Optional["ExecutorDefinition"] = None,
@@ -601,7 +601,7 @@ def to_job(
 
     def execute_in_process(
         self,
-        run_config: Any = None,
+        run_config: Optional[Any] = None,
         instance: Optional["DagsterInstance"] = None,
         resources: Optional[Dict[str, Any]] = None,
         raise_on_error: bool = True,

python_modules/dagster/dagster/core/definitions/config.py

@@ -52,7 +52,7 @@ class ConfigMapping(
     def __new__(
         cls,
         config_fn: Callable[[Any], Any],
-        config_schema: Any = None,
+        config_schema: Optional[Any] = None,
         receive_processed_config_values: Optional[bool] = None,
     ):
         return super(ConfigMapping, cls).__new__(

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

@@ -17,7 +17,7 @@ def __init__(
         input_defs: Optional[List[InputDefinition]] = None,
         output_defs: Optional[List[OutputDefinition]] = None,
         description: Optional[str] = None,
-        config_schema: Any = None,
+        config_schema: Optional[Any] = None,
         config_fn: Optional[Callable[[dict], dict]] = None,
     ):
         self.name = check.opt_str_param(name, "name")
@@ -89,7 +89,7 @@ def composite_solid(
 
 
 def composite_solid(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     input_defs: Optional[List[InputDefinition]] = None,
     output_defs: Optional[List[OutputDefinition]] = None,
     description: Optional[str] = None,

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

@@ -109,7 +109,7 @@ def graph(
 
 
 def graph(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     description: Optional[str] = None,
     input_defs: Optional[List[InputDefinition]] = None,
     output_defs: Optional[List[OutputDefinition]] = None,

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

@@ -78,7 +78,7 @@ def event_list_hook(
 
 
 def event_list_hook(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     required_resource_keys: Optional[AbstractSet[str]] = None,
     decorated_fn: Optional[Callable[..., Any]] = None,
 ) -> Union[HookDefinition, _Hook]:
@@ -145,7 +145,7 @@ def success_hook(
 
 
 def success_hook(
-    name: Union[SuccessOrFailureHookFn, Optional[str]] = None,
+    name: Optional[Union[SuccessOrFailureHookFn, str]] = None,
     required_resource_keys: Optional[AbstractSet[str]] = None,
 ) -> Union[HookDefinition, _Hook, Callable[[SuccessOrFailureHookFn], Union[HookDefinition, _Hook]]]:
     """Create a hook on step success events with the specified parameters from the decorated function.
@@ -219,7 +219,7 @@ def failure_hook(
 
 
 def failure_hook(
-    name: Union[SuccessOrFailureHookFn, Optional[str]] = None,
+    name: Optional[Union[SuccessOrFailureHookFn, str]] = None,
     required_resource_keys: Optional[AbstractSet[str]] = None,
 ) -> Union[HookDefinition, _Hook, Callable[[SuccessOrFailureHookFn], Union[HookDefinition, _Hook]]]:
     """Create a hook on step failure events with the specified parameters from the decorated function.

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

@@ -25,7 +25,7 @@ def __init__(
         description: Optional[str] = None,
         tags: Optional[Dict[str, Any]] = None,
         resource_defs: Optional[Dict[str, ResourceDefinition]] = None,
-        config: Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"] = None,
+        config: Optional[Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"]] = None,
         logger_defs: Optional[Dict[str, LoggerDefinition]] = None,
         executor_def: Optional["ExecutorDefinition"] = None,
         hooks: Optional[AbstractSet[HookDefinition]] = None,
@@ -120,10 +120,10 @@ def job(
 
 
 def job(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     description: Optional[str] = None,
     resource_defs: Optional[Dict[str, ResourceDefinition]] = None,
-    config: Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"] = None,
+    config: Optional[Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"]] = None,
     tags: Optional[Dict[str, Any]] = None,
     logger_defs: Optional[Dict[str, LoggerDefinition]] = None,
     executor_def: Optional["ExecutorDefinition"] = None,

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

@@ -199,7 +199,7 @@ def op(
 
 
 def op(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     description: Optional[str] = None,
     ins: Optional[Dict[str, In]] = None,
     out: Optional[Union[Out, Dict[str, Out]]] = None,

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

@@ -133,7 +133,7 @@ def pipeline(
 
 
 def pipeline(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     description: Optional[str] = None,
     mode_defs: Optional[List[ModeDefinition]] = None,
     preset_defs: Optional[List[PresetDefinition]] = None,

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

@@ -108,7 +108,7 @@ def repository(name: Optional[str] = ..., description: Optional[str] = ...) -> _
 
 
 def repository(
-    name: Union[Optional[str], Callable[..., Any]] = None, description: Optional[str] = None
+    name: Optional[Union[str, Callable[..., Any]]] = None, description: Optional[str] = None
 ) -> Union[RepositoryDefinition, _Repository]:
     """Create a repository from the decorated function.
 

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

@@ -155,7 +155,7 @@ def solid(
 
 
 def solid(
-    name: Union[Callable[..., Any], Optional[str]] = None,
+    name: Optional[Union[Callable[..., Any], str]] = None,
     description: Optional[str] = None,
     input_defs: Optional[Sequence[InputDefinition]] = None,
     output_defs: Optional[Sequence[OutputDefinition]] = None,
@@ -405,7 +405,7 @@ def is_context_provided(params: List[funcsigs.Parameter]) -> bool:
 
 
 def lambda_solid(
-    name: Union[Optional[str], Callable[..., Any]] = None,
+    name: Optional[Union[str, Callable[..., Any]]] = None,
     description: Optional[str] = None,
     input_defs: Optional[List[InputDefinition]] = None,
     output_def: Optional[OutputDefinition] = None,

python_modules/dagster/dagster/core/definitions/dependency.py

@@ -80,8 +80,8 @@ def __new__(
         cls,
         name: str,
         alias: Optional[str] = None,
-        tags: Dict[str, str] = None,
-        hook_defs: AbstractSet[HookDefinition] = None,
+        tags: Optional[Dict[str, str]] = None,
+        hook_defs: Optional[AbstractSet[HookDefinition]] = None,
         retry_policy: Optional[RetryPolicy] = None,
     ):
         return super().__new__(
@@ -109,7 +109,7 @@ def __init__(
         name: str,
         definition: "NodeDefinition",
         graph_definition: "GraphDefinition",
-        tags: Dict[str, str] = None,
+        tags: Optional[Dict[str, str]] = None,
         hook_defs: Optional[AbstractSet[HookDefinition]] = None,
         retry_policy: Optional[RetryPolicy] = None,
     ):

python_modules/dagster/dagster/core/definitions/events.py

@@ -468,7 +468,7 @@ class Materialization(
 
     def __new__(
         cls,
-        label: str = None,
+        label: Optional[str] = None,
         description: Optional[str] = None,
         metadata_entries: Optional[List[MetadataEntry]] = None,
         asset_key: Optional[Union[str, AssetKey]] = None,

python_modules/dagster/dagster/core/definitions/graph_definition.py

@@ -452,7 +452,7 @@ def to_job(
         name: Optional[str] = None,
         description: Optional[str] = None,
         resource_defs: Optional[Dict[str, ResourceDefinition]] = None,
-        config: Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"] = None,
+        config: Optional[Union[ConfigMapping, Dict[str, Any], "PartitionedConfig"]] = None,
         tags: Optional[Dict[str, Any]] = None,
         logger_defs: Optional[Dict[str, LoggerDefinition]] = None,
         executor_def: Optional["ExecutorDefinition"] = None,

python_modules/dagster/dagster/core/definitions/job_definition.py

@@ -62,7 +62,7 @@ def __init__(
         name: Optional[str] = None,
         description: Optional[str] = None,
         preset_defs: Optional[List[PresetDefinition]] = None,
-        tags: Dict[str, Any] = None,
+        tags: Optional[Dict[str, Any]] = None,
         hook_defs: Optional[AbstractSet[HookDefinition]] = None,
         op_retry_policy: Optional[RetryPolicy] = None,
         version_strategy: Optional[VersionStrategy] = None,

python_modules/dagster/dagster/core/definitions/pipeline_definition.py

@@ -150,7 +150,7 @@ def __init__(
         ] = None,
         mode_defs: Optional[List[ModeDefinition]] = None,
         preset_defs: Optional[List[PresetDefinition]] = None,
-        tags: Dict[str, Any] = None,
+        tags: Optional[Dict[str, Any]] = None,
         hook_defs: Optional[AbstractSet[HookDefinition]] = None,
         solid_retry_policy: Optional[RetryPolicy] = None,
         graph_def=None,

python_modules/dagster/dagster/core/definitions/preset.py

@@ -57,7 +57,7 @@ def __new__(
         run_config: Optional[Dict[str, object]] = None,
         solid_selection: Optional[List[str]] = None,
         mode: Optional[str] = None,
-        tags: Dict[str, object] = None,
+        tags: Optional[Dict[str, object]] = None,
     ):
 
         return super(PresetDefinition, cls).__new__(

python_modules/dagster/dagster/core/definitions/run_config.py

@@ -52,7 +52,9 @@ def remove_none_entries(ddict: dict) -> dict:
     return {k: v for k, v in ddict.items() if v is not None}
 
 
-def def_config_field(configurable_def: ConfigurableDefinition, is_required: bool = None) -> Field:
+def def_config_field(
+    configurable_def: ConfigurableDefinition, is_required: Optional[bool] = None
+) -> Field:
     return Field(
         Shape(
             {"config": configurable_def.config_field} if configurable_def.has_config_field else {}