[refactor] remove frozenlist

python_modules/dagster/dagster/_check/__init__.py

@@ -22,6 +22,8 @@
     overload,
 )
 
+from typing_extensions import Literal
+
 TypeOrTupleOfTypes = Union[type, Tuple[type, ...]]
 Numeric = Union[int, float]
 T = TypeVar("T")
@@ -768,12 +770,8 @@ def list_param(
     of_type: Optional[TypeOrTupleOfTypes] = None,
     additional_message: Optional[str] = None,
 ) -> List[Any]:
-    from dagster._utils import frozenlist
-
-    if not isinstance(obj, (frozenlist, list)):
-        raise _param_type_mismatch_exception(
-            obj, (frozenlist, list), param_name, additional_message
-        )
+    if not isinstance(obj, list):
+        raise _param_type_mismatch_exception(obj, list, param_name, additional_message)
 
     if not of_type:
         return obj
@@ -793,12 +791,8 @@ def opt_list_param(
     If the of_type argument is provided, also ensures that list items conform to the type specified
     by of_type.
     """
-    from dagster._utils import frozenlist
-
-    if obj is not None and not isinstance(obj, (frozenlist, list)):
-        raise _param_type_mismatch_exception(
-            obj, (frozenlist, list), param_name, additional_message
-        )
+    if obj is not None and not isinstance(obj, list):
+        raise _param_type_mismatch_exception(obj, list, param_name, additional_message)
 
     if not obj:
         return []
@@ -840,12 +834,8 @@ def opt_nullable_list_param(
     If the of_type argument is provided, also ensures that list items conform to the type specified
     by of_type.
     """
-    from dagster._utils import frozenlist
-
-    if obj is not None and not isinstance(obj, (frozenlist, list)):
-        raise _param_type_mismatch_exception(
-            obj, (frozenlist, list), param_name, additional_message
-        )
+    if obj is not None and not isinstance(obj, list):
+        raise _param_type_mismatch_exception(obj, list, param_name, additional_message)
 
     if not obj:
         return None if obj is None else []
@@ -1180,39 +1170,88 @@ def opt_path_param(
 # ########################
 
 
+@overload
+def sequence_param(
+    obj: Sequence[T],
+    param_name: str,
+    of_type: Optional[TypeOrTupleOfTypes] = ...,
+    additional_message: Optional[str] = ...,
+    as_tuple: Literal[True] = ...,
+) -> Tuple[T, ...]:
+    ...
+
+
+@overload
 def sequence_param(
     obj: Sequence[T],
     param_name: str,
     of_type: Optional[TypeOrTupleOfTypes] = None,
     additional_message: Optional[str] = None,
+    as_tuple: bool = ...,
 ) -> Sequence[T]:
+    ...
+
+
+def sequence_param(
+    obj: Sequence[T],
+    param_name: str,
+    of_type: Optional[TypeOrTupleOfTypes] = None,
+    additional_message: Optional[str] = None,
+    as_tuple: bool = False,
+) -> Union[Sequence[T], Tuple[T, ...]]:
     if not isinstance(obj, collections.abc.Sequence):
         raise _param_type_mismatch_exception(
             obj, (collections.abc.Sequence,), param_name, additional_message
         )
 
+    norm_obj = tuple(obj) if as_tuple else obj
     if not of_type:
-        return obj
+        return norm_obj
+
+    return _check_iterable_items(norm_obj, of_type, "sequence")
 
-    return _check_iterable_items(obj, of_type, "sequence")
 
+@overload
+def opt_sequence_param(
+    obj: Optional[Sequence[T]],
+    param_name: str,
+    of_type: Optional[TypeOrTupleOfTypes] = ...,
+    additional_message: Optional[str] = ...,
+    as_tuple: Literal[True] = ...,
+) -> Tuple[T, ...]:
+    ...
 
+
+@overload
 def opt_sequence_param(
     obj: Optional[Sequence[T]],
     param_name: str,
     of_type: Optional[TypeOrTupleOfTypes] = None,
     additional_message: Optional[str] = None,
+    as_tuple: bool = ...,
 ) -> Sequence[T]:
+    ...
+
+
+def opt_sequence_param(
+    obj: Optional[Sequence[T]],
+    param_name: str,
+    of_type: Optional[TypeOrTupleOfTypes] = None,
+    additional_message: Optional[str] = None,
+    as_tuple: bool = False,
+) -> Union[Sequence[T], Tuple[T, ...]]:
     if obj is None:
-        return []
+        return tuple() if as_tuple else []
     elif not isinstance(obj, collections.abc.Sequence):
         raise _param_type_mismatch_exception(
             obj, (collections.abc.Sequence,), param_name, additional_message
         )
-    elif of_type is not None:
-        return _check_iterable_items(obj, of_type, "sequence")
+
+    norm_obj = tuple(obj) if as_tuple else obj
+    if of_type is not None:
+        return _check_iterable_items(norm_obj, of_type, "sequence")
     else:
-        return obj
+        return norm_obj
 
 
 @overload
@@ -1221,16 +1260,29 @@ def opt_nullable_sequence_param(
     param_name: str,
     of_type: Optional[TypeOrTupleOfTypes] = ...,
     additional_message: Optional[str] = ...,
+    as_tuple: bool = ...,
 ) -> None:
     ...
 
 
+@overload
+def opt_nullable_sequence_param(
+    obj: Sequence[T],
+    param_name: str,
+    of_type: Optional[TypeOrTupleOfTypes] = ...,
+    additional_message: Optional[str] = ...,
+    as_tuple: Literal[True] = ...,
+) -> Tuple[T, ...]:
+    ...
+
+
 @overload
 def opt_nullable_sequence_param(
     obj: Sequence[T],
     param_name: str,
     of_type: Optional[TypeOrTupleOfTypes] = None,
     additional_message: Optional[str] = ...,
+    as_tuple: bool = ...,
 ) -> Sequence[T]:
     ...
 
@@ -1240,11 +1292,12 @@ def opt_nullable_sequence_param(
     param_name: str,
     of_type: Optional[TypeOrTupleOfTypes] = None,
     additional_message: Optional[str] = None,
+    as_tuple: bool = False,
 ) -> Optional[Sequence[T]]:
     if obj is None:
         return None
     else:
-        return opt_sequence_param(obj, param_name, of_type, additional_message)
+        return opt_sequence_param(obj, param_name, of_type, additional_message, as_tuple)
 
 
 # ########################

python_modules/dagster/dagster/_config/post_process.py

@@ -2,7 +2,7 @@
 from typing import Any, Dict, List, Mapping, Optional, cast
 
 import dagster._check as check
-from dagster._utils import ensure_single_item, frozendict, frozenlist
+from dagster._utils import ensure_single_item, frozendict
 from dagster._utils.error import serializable_error_info_from_exc_info
 
 from .config_type import ConfigType, ConfigTypeKind
@@ -210,7 +210,7 @@ def _recurse_in_to_array(context: TraversalContext, config_value: Any) -> Evalua
     if errors:
         return EvaluateValueResult.for_errors(errors)
 
-    return EvaluateValueResult.for_value(frozenlist([result.value for result in results]))
+    return EvaluateValueResult.for_value([result.value for result in results])
 
 
 def _recurse_in_to_map(context: TraversalContext, config_value: Any) -> EvaluateValueResult[Any]:

python_modules/dagster/dagster/_core/code_pointer.py

@@ -4,13 +4,13 @@
 import sys
 from abc import ABC, abstractmethod
 from types import ModuleType
-from typing import Callable, List, NamedTuple, Optional, Sequence, cast
+from typing import Callable, List, NamedTuple, Optional, Sequence, Tuple, cast
 
 import dagster._check as check
 from dagster._core.errors import DagsterImportError, DagsterInvariantViolationError
 from dagster._serdes import whitelist_for_serdes
 from dagster._seven import get_import_error_message, import_module_from_path
-from dagster._utils import alter_sys_path, frozenlist
+from dagster._utils import alter_sys_path
 
 
 class CodePointer(ABC):
@@ -269,8 +269,8 @@ class CustomPointer(
         "_CustomPointer",
         [
             ("reconstructor_pointer", ModuleCodePointer),
-            ("reconstructable_args", Sequence[object]),
-            ("reconstructable_kwargs", Sequence[Sequence]),
+            ("reconstructable_args", Tuple[object, ...]),
+            ("reconstructable_kwargs", Tuple[Tuple[str, object], ...]),
         ],
     ),
     CodePointer,
@@ -298,9 +298,9 @@ def __new__(
 
         # These are frozenlists, rather than lists, so that they can be hashed and the pointer
         # stored in the lru_cache on the repository and pipeline get_definition methods
-        reconstructable_args = frozenlist(reconstructable_args)
-        reconstructable_kwargs = frozenlist(
-            [frozenlist(reconstructable_kwarg) for reconstructable_kwarg in reconstructable_kwargs]
+        reconstructable_args = tuple(reconstructable_args)
+        reconstructable_kwargs = tuple(
+            tuple(reconstructable_kwarg) for reconstructable_kwarg in reconstructable_kwargs
         )
 
         return super(CustomPointer, cls).__new__(

python_modules/dagster/dagster/_core/definitions/cacheable_assets.py

@@ -2,7 +2,7 @@
 import inspect
 import json
 from abc import ABC, abstractmethod
-from typing import AbstractSet, Any, List, Mapping, NamedTuple, Optional, Sequence
+from typing import AbstractSet, Any, List, Mapping, NamedTuple, Optional, Sequence, Tuple
 
 import dagster._check as check
 import dagster._seven as seven
@@ -14,7 +14,7 @@
 from dagster._core.definitions.resource_definition import ResourceDefinition
 from dagster._core.definitions.resource_requirement import ResourceAddable
 from dagster._serdes import whitelist_for_serdes
-from dagster._utils import frozendict, frozenlist, make_readonly_value
+from dagster._utils import frozendict, make_readonly_value
 
 
 @whitelist_for_serdes
@@ -27,7 +27,7 @@ class AssetsDefinitionCacheableData(
             ("internal_asset_deps", Optional[Mapping[str, AbstractSet[AssetKey]]]),
             ("group_name", Optional[str]),
             ("metadata_by_output_name", Optional[Mapping[str, MetadataUserInput]]),
-            ("key_prefix", Optional[CoercibleToAssetKeyPrefix]),
+            ("key_prefix", Optional[Tuple[str, ...]]),
             ("can_subset", bool),
             ("extra_metadata", Optional[Mapping[Any, Any]]),
             ("freshness_policies_by_output_name", Optional[Mapping[str, FreshnessPolicy]]),
@@ -95,7 +95,7 @@ def __new__(
             metadata_by_output_name=make_readonly_value(metadata_by_output_name)
             if metadata_by_output_name
             else None,
-            key_prefix=frozenlist(key_prefix) if key_prefix else None,
+            key_prefix=tuple(key_prefix) if key_prefix else None,
             can_subset=check.opt_bool_param(can_subset, "can_subset", default=False),
             extra_metadata=make_readonly_value(extra_metadata) if extra_metadata else None,
             freshness_policies_by_output_name=frozendict(freshness_policies_by_output_name)

python_modules/dagster/dagster/_core/definitions/metadata/table.py

@@ -5,7 +5,6 @@
 from dagster._serdes.serdes import (
     whitelist_for_serdes,
 )
-from dagster._utils import frozenlist
 
 # ########################
 # ##### TABLE RECORD
@@ -109,7 +108,7 @@ def __new__(
     ):
         return super(TableSchema, cls).__new__(
             cls,
-            columns=frozenlist(check.sequence_param(columns, "columns", of_type=TableColumn)),
+            columns=check.sequence_param(columns, "columns", of_type=TableColumn),
             constraints=check.opt_inst_param(
                 constraints, "constraints", TableConstraints, default=_DEFAULT_TABLE_CONSTRAINTS
             ),
@@ -157,7 +156,7 @@ def __new__(
     ):
         return super(TableConstraints, cls).__new__(
             cls,
-            other=frozenlist(check.sequence_param(other, "other", of_type=str)),
+            other=check.sequence_param(other, "other", of_type=str),
         )
 
 
@@ -255,7 +254,7 @@ def __new__(
             cls,
             nullable=check.bool_param(nullable, "nullable"),
             unique=check.bool_param(unique, "unique"),
-            other=frozenlist(check.opt_sequence_param(other, "other")),
+            other=check.opt_sequence_param(other, "other"),
         )
 
 

python_modules/dagster/dagster/_core/definitions/node_definition.py

@@ -13,7 +13,7 @@
 import dagster._check as check
 from dagster._core.definitions.configurable import NamedConfigurableDefinition
 from dagster._core.definitions.policy import RetryPolicy
-from dagster._utils import frozendict, frozenlist
+from dagster._utils import frozendict
 
 from .hook_definition import HookDefinition
 from .utils import check_valid_name, validate_tags
@@ -54,10 +54,10 @@ def __init__(
         self._name = check_valid_name(name)
         self._description = check.opt_str_param(description, "description")
         self._tags = validate_tags(tags)
-        self._input_defs = frozenlist(input_defs)
+        self._input_defs = tuple(input_defs)
         self._input_dict = frozendict({input_def.name: input_def for input_def in input_defs})
         check.invariant(len(self._input_defs) == len(self._input_dict), "Duplicate input def names")
-        self._output_defs = frozenlist(output_defs)
+        self._output_defs = tuple(output_defs)
         self._output_dict = frozendict({output_def.name: output_def for output_def in output_defs})
         check.invariant(
             len(self._output_defs) == len(self._output_dict), "Duplicate output def names"

python_modules/dagster/dagster/_core/definitions/partition.py

@@ -38,7 +38,6 @@
 from dagster._core.storage.tags import PARTITION_NAME_TAG
 from dagster._serdes import whitelist_for_serdes
 from dagster._seven.compat.pendulum import PendulumDateTime, to_timezone
-from dagster._utils import frozenlist
 from dagster._utils.backcompat import deprecation_warning, experimental_arg_warning
 from dagster._utils.merger import merge_dicts
 from dagster._utils.schedules import schedule_execution_time_iterator
@@ -949,9 +948,7 @@ def _execution_fn(context):
                 return
 
             selected_partitions = (
-                selector_result
-                if isinstance(selector_result, (frozenlist, list))
-                else [selector_result]
+                selector_result if isinstance(selector_result, list) else [selector_result]
             )
 
             check.is_list(selected_partitions, of_type=Partition)

python_modules/dagster/dagster/_core/definitions/reconstruct.py

@@ -40,7 +40,7 @@
 )
 from dagster._core.selector import parse_solid_selection
 from dagster._serdes import pack_value, unpack_value, whitelist_for_serdes
-from dagster._utils import frozenlist, make_readonly_value
+from dagster._utils import make_readonly_value
 
 from .events import AssetKey
 from .pipeline_base import IPipeline
@@ -71,7 +71,7 @@ class ReconstructableRepository(
             ("pointer", CodePointer),
             ("container_image", Optional[str]),
             ("executable_path", Optional[str]),
-            ("entry_point", Sequence[str]),
+            ("entry_point", Tuple[str, ...]),
             ("container_context", Optional[Mapping[str, Any]]),
             ("repository_load_data", Optional["RepositoryLoadData"]),
         ],
@@ -94,7 +94,7 @@ def __new__(
             container_image=check.opt_str_param(container_image, "container_image"),
             executable_path=check.opt_str_param(executable_path, "executable_path"),
             entry_point=(
-                frozenlist(check.sequence_param(entry_point, "entry_point", of_type=str))
+                tuple(check.sequence_param(entry_point, "entry_point", of_type=str))
                 if entry_point is not None
                 else DEFAULT_DAGSTER_ENTRY_POINT
             ),