refactor: Update type imports to use collections.abc for better clarity and consistency

Author: thorwhalen

meshed/base.py

@@ -5,7 +5,8 @@
 from collections import Counter
 from dataclasses import dataclass, field, fields
 from functools import partial, cached_property
-from typing import Callable, MutableMapping, Iterable, Union, Sized, Sequence, Literal
+from typing import Union, Literal
+from collections.abc import Callable, MutableMapping, Iterable, Sized, Sequence
 
 from i2 import Sig, call_somewhat_forgivingly
 from i2.signatures import (
@@ -782,7 +783,7 @@ def _mapped_extraction(src: dict, to_extract: dict):
             yield desired_name, src[src_name]
 
 
-def duplicates(elements: Union[Iterable, Sized]):
+def duplicates(elements: Iterable | Sized):
     c = Counter(elements)
     if len(c) != len(elements):
         return [name for name, count in c.items() if count > 1]
@@ -835,17 +836,18 @@ def _complete_dict_with_iterable_of_required_keys(
             to_complete[required_key] = required_key
 
 
-from typing import NewType, Dict, Tuple, Mapping
+from typing import NewType, Dict, Tuple
+from collections.abc import Mapping
 
 # TODO: Make a type where ``isinstance(s, Identifier) == s.isidentifier()``
 Identifier = NewType("Identifier", str)  # + should satisfy str.isidentifier
 Bind = Union[
     str,  # Identifier or ' '.join(Iterable[Identifier])
-    Dict[Identifier, Identifier],
-    Sequence[Union[Identifier, Tuple[Identifier, Identifier]]],
+    dict[Identifier, Identifier],
+    Sequence[Union[Identifier, tuple[Identifier, Identifier]]],
 ]
 
-IdentifierMapping = Dict[Identifier, Identifier]
+IdentifierMapping = dict[Identifier, Identifier]
 
 
 def identifier_mapping(x: Bind) -> IdentifierMapping:

meshed/components.py

@@ -3,7 +3,8 @@
 """
 
 from i2 import Sig
-from typing import Callable, Any
+from typing import Any
+from collections.abc import Callable
 from operator import itemgetter, attrgetter
 from dataclasses import dataclass
 from functools import partial

meshed/composition.py

@@ -2,7 +2,8 @@
 
 from dataclasses import fields
 from inspect import signature
-from typing import Callable, Union
+from typing import Union
+from collections.abc import Callable
 from functools import partial
 
 from i2 import Sig, kwargs_trans
@@ -38,7 +39,7 @@ def func_node_kwargs_trans(func: Callable) -> FuncNodeKwargsTrans:
 
 
 def func_node_name_trans(
-    name_trans: Callable[[str], Union[str, None]],
+    name_trans: Callable[[str], str | None],
     *,
     also_apply_to_func_label: bool = False,
 ):
@@ -68,7 +69,7 @@ def name_trans(name):
 # TODO: Extract  ingress/egress boilerplate to wrapper
 def suffix_ids(
     func_nodes,
-    renamer: Union[Renamer, str] = numbered_suffix_renamer,
+    renamer: Renamer | str = numbered_suffix_renamer,
     *,
     also_apply_to_func_label: bool = False,
 ):

meshed/dag.py

@@ -138,17 +138,14 @@
 from itertools import chain
 from operator import attrgetter, eq
 from typing import (
-    Callable,
-    MutableMapping,
     Union,
     Optional,
-    Iterable,
     Any,
-    Mapping,
     Tuple,
     KT,
     VT,
 )
+from collections.abc import Callable, MutableMapping, Iterable, Mapping
 from warnings import warn
 
 from i2 import double_up_as_factory, MultiFunc
@@ -201,7 +198,7 @@
 
 from meshed.viz import dot_lines_of_objs, add_new_line_if_none
 
-FuncMapping = Union[Mapping[KT, Callable], Iterable[Tuple[KT, Callable]]]
+FuncMapping = Union[Mapping[KT, Callable], Iterable[tuple[KT, Callable]]]
 
 
 def order_subset_from_list(items, sublist):
@@ -490,7 +487,7 @@ class DAG:
 
     """
 
-    func_nodes: Iterable[Union[FuncNode, Callable]] = ()
+    func_nodes: Iterable[FuncNode | Callable] = ()
     cache_last_scope: bool = field(default=True, repr=False)
     parameter_merge: ParameterMerger = field(
         default=conservative_parameter_merge, repr=False
@@ -1163,7 +1160,7 @@ def params_for_src(self):
                 d[src_name].append(node.sig.parameters[arg_name])
         return dict(d)
 
-    def src_name_params(self, src_names: Optional[Iterable[str]] = None):
+    def src_name_params(self, src_names: Iterable[str] | None = None):
         """Generate Parameter instances that are needed to compute ``src_names``"""
         # see params_for_src property to see what d is
         d = self.params_for_src
@@ -1613,7 +1610,7 @@ def dag():
     return func_nodes_to_code(dag.func_nodes, dag.name)
 
 
-def parametrized_dag_factory(dag: DAG, param_var_nodes: Union[str, Iterable[str]]):
+def parametrized_dag_factory(dag: DAG, param_var_nodes: str | Iterable[str]):
     """
     Constructs a factory for sub-DAGs derived from the input DAG, with values of
     specific 'parameter' variable nodes precomputed and fixed. These precomputed nodes,
@@ -1794,7 +1791,7 @@ def ch_funcs(
     func_nodes: DagAble = None,
     *,
     func_mapping: FuncMapping = (),
-    validate_func_mapping: Optional[FuncMappingValidator] = _validate_func_mapping,
+    validate_func_mapping: FuncMappingValidator | None = _validate_func_mapping,
     # TODO: Design. Don't like the fact that ch_func_node_func needs a slot for
     #  func_comparator, which is then given later. Perhaps only ch_func_node_func should
     #  should be given (and it contains the func_comparator)

meshed/ext/gk.py

@@ -17,7 +17,7 @@
 # ---------- base --------------------------------------------------------------
 
 
-class Data(object):
+class Data:
     """
     This wraps any data that is consumed or produced
     by a Operation. This data should also know how to serialize
@@ -145,7 +145,7 @@ def __repr__(self):
         """
         Display more informative names for the Operation class
         """
-        return "%s(name='%s', needs=%s, provides=%s)" % (
+        return "{}(name='{}', needs={}, provides={})".format(
             self.__class__.__name__,
             self.name,
             self.needs,
@@ -269,7 +269,7 @@ def __repr__(self):
         return 'DeleteInstruction("%s")' % self
 
 
-class Network(object):
+class Network:
     """
     This is the main network implementation. The class contains all of the
     code necessary to weave together operations into a directed-acyclic-graph (DAG)
@@ -878,7 +878,7 @@ def __repr__(self):
         """
         Display more informative names for the Operation class
         """
-        return "%s(name='%s', needs=%s, provides=%s, fn=%s)" % (
+        return "{}(name='{}', needs={}, provides={}, fn={})".format(
             self.__class__.__name__,
             self.name,
             self.needs,
@@ -887,7 +887,7 @@ def __repr__(self):
         )
 
 
-class compose(object):
+class compose:
     """
     This is a simple class that's used to compose ``operation`` instances into
     a computation graph.

meshed/itools.py

@@ -5,16 +5,12 @@
 
 from typing import (
     Any,
-    Mapping,
-    Sized,
-    MutableMapping,
-    Iterable,
-    Callable,
     List,
     TypeVar,
     Union,
     Optional,
 )
+from collections.abc import Mapping, Sized, MutableMapping, Iterable, Callable
 from itertools import product, chain
 from functools import wraps, reduce, partial
 from collections import defaultdict
@@ -28,7 +24,7 @@
 MutableGraph = MutableMapping[N, Iterable[N]]
 
 
-def _import_or_raise(module_name, pip_install_name: Optional[Union[str, bool]] = None):
+def _import_or_raise(module_name, pip_install_name: str | bool | None = None):
     try:
         return __import__(module_name)
     except ImportError as e:
@@ -345,7 +341,7 @@ def root_nodes(g: Graph):
 
 
 # TODO: Can be made much more efficient, by looking at the ancestors code itself
-def root_ancestors(graph: dict, nodes: Union[str, Iterable[str]]):
+def root_ancestors(graph: dict, nodes: str | Iterable[str]):
     """
     Returns the roots of the sub-dag that contribute to compute the given nodes.
     """
@@ -425,7 +421,7 @@ def reverse_edges(g: Graph):
         yield from product(dst_nodes, src)
 
 
-def has_cycle(g: Graph) -> List[N]:
+def has_cycle(g: Graph) -> list[N]:
     """
         Returns a list representing a cycle in the graph if any. An empty list indicates no cycle.
 
@@ -653,7 +649,7 @@ def filter_dict_on_keys(d, condition):
 
 
 def nodes_of_graph(graph):
-    return set([*graph.keys(), *graph.values()])
+    return {*graph.keys(), *graph.values()}
 
 
 def subtract_subgraph(graph, subgraph):

meshed/makers.py

@@ -149,14 +149,11 @@
 from typing import (
     Tuple,
     Optional,
-    Iterator,
-    Iterable,
     TypeVar,
-    Callable,
     Dict,
-    Mapping,
     Union,
 )
+from collections.abc import Iterator, Iterable, Callable, Mapping
 from functools import partial
 
 
@@ -249,7 +246,7 @@ def parse_body(body, *, body_index=None):
 
 
 # Note: generalize? glom?
-def parse_assignment(body: ast.Assign, info=None) -> Tuple:
+def parse_assignment(body: ast.Assign, info=None) -> tuple:
     # TODO: Make this validation better (at least more help in raised error)
     # TODO: extract validating code out as validation functions?
     info = _ast_info_str(body)
@@ -430,7 +427,7 @@ def parse_steps(src):
 
 def src_to_func_node_factory(
     src, exclude_names=None
-) -> Iterator[Union[FuncNode, FuncNodeFactory]]:
+) -> Iterator[FuncNode | FuncNodeFactory]:
     """
     :param src: Callable or string of callable.
     :param exclude_names: Names to exclude when making func_nodes
@@ -454,7 +451,7 @@ def src_to_func_node_factory(
 #  to make itself. Do we gain much over simply saying "factory, make yourself"?
 def dlft_factory_to_func(
     factory: partial,
-    name_to_func_map: Optional[Dict[str, Callable]] = None,
+    name_to_func_map: dict[str, Callable] | None = None,
     use_place_holder_fallback=True,
 ):
     """Get a function for the given factory, using"""
@@ -545,7 +542,7 @@ def code_to_fnodes(
     *,
     func_src: FuncSource = dlft_factory_to_func,
     use_place_holder_fallback=False,
-) -> Tuple[FuncNode]:
+) -> tuple[FuncNode]:
     """Get func_nodes from src code"""
     func_src = _ensure_func_src(func_src, use_place_holder_fallback)
     # Pass on to _code_to_fnodes to get func nodes iterable needed to make DAG
@@ -593,10 +590,10 @@ def code_to_digraph(src):
 # from typing import Tuple, Iterable, Iterator
 # from meshed import FuncNode, code_to_dag, code_to_fnodes, DAG
 
-extract_tokens = re.compile("\w+").findall
+extract_tokens = re.compile(r"\w+").findall
 
 
-def triples_to_fnodes(triples: Iterable[Tuple[str, str, str]]) -> Iterable[FuncNode]:
+def triples_to_fnodes(triples: Iterable[tuple[str, str, str]]) -> Iterable[FuncNode]:
     """Converts an iterable of func call triples to an iterable of ``FuncNode``s.
     (Which in turn can be converted to a ``DAG``.)
 
@@ -624,9 +621,9 @@ def triples_to_fnodes(triples: Iterable[Tuple[str, str, str]]) -> Iterable[FuncN
 
 
 def _triple_to_func_call_str(
-    outputs: Union[str, Iterable[str]],
+    outputs: str | Iterable[str],
     func_name: str,
-    inputs: Union[str, Iterable[str]],
+    inputs: str | Iterable[str],
 ) -> str:
     """Converts a `(outputs, func_name, inputs)` triple to a function call string.
 
@@ -699,7 +696,7 @@ def lined_dag(funcs):
         return dag
 
 
-from typing import Callable, Mapping, Iterable
+from collections.abc import Callable, Mapping, Iterable
 
 NamedFuncs = Mapping[str, Callable]
 

meshed/scrap/annotations_to_meshes.py

@@ -7,7 +7,8 @@
 """
 
 import typing
-from typing import Dict, Protocol, Callable, TypeVar, Any
+from typing import Dict, Protocol, TypeVar, Any
+from collections.abc import Callable
 from collections.abc import Callable as CallableGenericAlias
 from inspect import signature, Signature, Parameter
 from functools import wraps
@@ -75,7 +76,7 @@ def callable_annots_to_signature(
 
 
 def func_types_to_protocol(
-    func_types: Dict[str, CallableGenericAlias],
+    func_types: dict[str, CallableGenericAlias],
     name: str = None,
     *,
     mk_argname: MkArgname = try_annotation_name,
@@ -102,7 +103,8 @@ def method_stub(*args, **kwargs):
 
 
 def test_func_types_to_protocol():
-    from typing import Iterable, Any, NewType, Callable
+    from typing import Any, NewType
+    from collections.abc import Iterable, Callable
 
     Group = NewType("Group", str)
     Item = NewType("Item", Any)
@@ -132,7 +134,7 @@ def items_for_group(self, group: Group) -> Iterable[Item]:
 
 
 def func_types_to_scaffold(
-    func_types: Dict[str, CallableGenericAlias], name: str = None
+    func_types: dict[str, CallableGenericAlias], name: str = None
 ) -> str:
     """Produces a scaffold class containing the said methods, with given annotations"""
 
@@ -181,7 +183,8 @@ def items_for_group(self, group: Group) -> Iterable:
 
 
 def test_func_types_to_scaffold():
-    from typing import Iterable, Any, NewType, Callable
+    from typing import Any, NewType
+    from collections.abc import Iterable, Callable
 
     Group = NewType("Group", str)
     Item = NewType("Item", Any)

meshed/scrap/cached_dag.py

@@ -1,4 +1,4 @@
-from typing import Mapping
+from collections.abc import Mapping
 from functools import partial
 from collections import ChainMap
 from meshed.itools import edge_reversed_graph, descendants