Prune unnecessary typing imports

Author: vkottler

vcorelib/args/__init__.py

@@ -5,21 +5,19 @@
 # built-in
 from argparse import ArgumentParser, Namespace
 from typing import Callable as _Callable
-from typing import Dict as _Dict
 from typing import Sequence as _Sequence
-from typing import Tuple as _Tuple
 
 CommandFunction = _Callable[[Namespace], int]
 CommandRegister = _Callable[[ArgumentParser], CommandFunction]
 
-CommandLoader = _Callable[[], _Sequence[_Tuple[str, str, CommandRegister]]]
+CommandLoader = _Callable[[], _Sequence[tuple[str, str, CommandRegister]]]
 
-CMDS: _Dict[str, CommandFunction] = {}
+CMDS: dict[str, CommandFunction] = {}
 
 
 def app_args(
-    command_loader: CommandLoader, commands: _Dict[str, CommandFunction] = None
-) -> _Tuple[_Callable[[ArgumentParser], None], CommandFunction]:
+    command_loader: CommandLoader, commands: dict[str, CommandFunction] = None
+) -> tuple[_Callable[[ArgumentParser], None], CommandFunction]:
     """
     Create a function that can be used to add sub-command processing to an
     argument parser.

vcorelib/asyncio/__init__.py

@@ -17,9 +17,7 @@
 from typing import Coroutine as _Coroutine
 from typing import Iterable as _Iterable
 from typing import Iterator
-from typing import List as _List
 from typing import Optional as _Optional
-from typing import Set as _Set
 from typing import TypeVar as _TypeVar
 
 # internal
@@ -50,7 +48,7 @@ def log_task_exception(
 
 def log_exceptions(
     tasks: _Iterable[_asyncio.Task[T]], logger: _LoggerType = None
-) -> _List[_asyncio.Task[T]]:
+) -> list[_asyncio.Task[T]]:
     """Log task exception and return the list of tasks that aren't complete."""
 
     for task in tasks:
@@ -147,7 +145,7 @@ def setter(sig: int, _: _Optional[_FrameType]) -> None:
     return setter
 
 
-def all_stop_signals() -> _Set[int]:
+def all_stop_signals() -> set[int]:
     """Get a set of all stop signals on this platform."""
 
     return {

vcorelib/asyncio/subprocess.py

@@ -5,7 +5,6 @@
 # built-in
 from asyncio import create_subprocess_exec
 from asyncio.subprocess import Process as _Process
-from typing import Tuple as _Tuple
 
 # internal
 from vcorelib.logging import LoggerType
@@ -16,7 +15,7 @@
 
 def log_process_info(
     program: str, *args: str, base: _Pathlike = None
-) -> _Tuple[str, str]:
+) -> tuple[str, str]:
     """
     Get a relative-path program string and a space-delimeted string of the
     arguments, which have also been shortened against a possible relative path.
@@ -34,7 +33,7 @@ async def create_subprocess_exec_log(
     *args: str,
     stdout: int = None,
     stderr: int = None,
-    rel: _Tuple[str, str] = None,
+    rel: tuple[str, str] = None,
     **kwargs,
 ) -> _Process:
     """

vcorelib/dict/__init__.py

@@ -8,16 +8,14 @@
 from enum import auto as _auto
 from logging import getLogger
 from typing import Any as _Any
-from typing import Dict as _Dict
 from typing import Iterator as _Iterator
-from typing import List as _List
 
 # internal
 from vcorelib.logging import LoggerType
 
 _LOG = getLogger(__name__)
-GenericDict = _Dict[_Any, _Any]
-GenericStrDict = _Dict[str, _Any]
+GenericDict = dict[_Any, _Any]
+GenericStrDict = dict[str, _Any]
 
 
 def consume(data: GenericDict, key: _Any, default: _Any = None) -> _Any:
@@ -81,7 +79,7 @@ class MergeStrategy(_Enum):
 def merge_recursive(
     dict_a: GenericDict,
     dict_b: GenericDict,
-    path: _List[str] = None,
+    path: list[str] = None,
     expect_overwrite: bool = False,
     logger: LoggerType = None,
     strategy: MergeStrategy = MergeStrategy.RECURSIVE,
@@ -140,7 +138,7 @@ def merge_recursive(
 def merge(
     dict_a: GenericDict,
     dict_b: GenericDict,
-    path: _List[str] = None,
+    path: list[str] = None,
     expect_overwrite: bool = False,
     logger: LoggerType = None,
     strategy: MergeStrategy = MergeStrategy.RECURSIVE,
@@ -162,7 +160,7 @@ def merge(
 
 
 def merge_dicts(
-    dicts: _List[GenericDict],
+    dicts: list[GenericDict],
     expect_overwrite: bool = False,
     logger: LoggerType = None,
     strategy: MergeStrategy = MergeStrategy.RECURSIVE,

vcorelib/dict/codec.py

@@ -7,12 +7,10 @@
 import abc as _abc
 from contextlib import contextmanager as _contextmanager
 from typing import Any as _Any
-from typing import Dict as _Dict
 from typing import Iterator as _Iterator
 from typing import Optional as _Optional
 from typing import Type as _Type
 from typing import TypeVar as _TypeVar
-from typing import Union as _Union
 
 # internal
 from vcorelib.io import ARBITER as _ARBITER
@@ -30,7 +28,7 @@ class JsonCodec(_abc.ABC, SchemaMixin):
     """A simple JSON codec interface."""
 
     @classmethod
-    def normalize(cls: _Type[T], data: _Union[_JsonObject, T]) -> T:
+    def normalize(cls: _Type[T], data: _JsonObject | T) -> T:
         """Ensure that some object is an instance of this class."""
 
         if not isinstance(data, cls):
@@ -112,7 +110,7 @@ def file_cache(
         cls: _Type[T],
         pathlike: _Pathlike,
         arbiter: _DataArbiter = _ARBITER,
-        encode_kwargs: _Dict[str, _Any] = None,
+        encode_kwargs: dict[str, _Any] = None,
         require_success: bool = False,
         **kwargs,
     ) -> _Iterator[T]:

vcorelib/dict/env.py

@@ -6,13 +6,12 @@
 # built-in
 from os import environ as _environ
 from typing import Any as _Any
-from typing import List as _List
 from typing import Mapping as _Mapping
 
 # internal
 from vcorelib.dict import GenericDict as _GenericDict
 
-GenericList = _List[_Any]
+GenericList = list[_Any]
 
 
 def str_resolve_env_var(data: str, env: _Mapping[str, _Any] = None) -> str:
@@ -74,7 +73,7 @@ def dict_resolve_env_vars(
     in-place.
     """
 
-    keys_to_remove: _List[str] = []
+    keys_to_remove: list[str] = []
     to_update: _GenericDict = {}
 
     for key, value in data.items():

vcorelib/graph/abc.py

@@ -4,9 +4,7 @@
 
 # built-in
 from collections import UserDict as _UserDict
-from typing import Dict as _Dict
 from typing import Iterator as _Iterator
-from typing import Set as _Set
 from typing import Type as _Type
 from typing import TypeVar as _TypeVar
 
@@ -90,11 +88,11 @@ def incoming(self, graph_kind: _Type[V] = None) -> _Iterator[T]:
                     yield graph[label]
                     break
 
-    def parallel(self) -> _Set[T]:
+    def parallel(self) -> set[T]:
         """Iterate over nodes that this instance shares parallel edges with."""
 
-        outgoing: _Set[T] = set(self.outgoing())
-        incoming: _Set[T] = set(self.incoming())
+        outgoing: set[T] = set(self.outgoing())
+        incoming: set[T] = set(self.incoming())
         return outgoing.intersection(incoming)
 
 
@@ -107,7 +105,7 @@ class AbstractDiGraph(
     def __init__(
         self,
         name: str,
-        initialdata: _Dict[str, T] = None,
+        initialdata: dict[str, T] = None,
         graph_attrs: _AttributeMap = None,
         node_attrs: _AttributeMap = None,
         edge_attrs: _AttributeMap = None,
@@ -121,7 +119,7 @@ def __init__(
         self.edge_attrs = edge_attrs
 
         # Source node -> a set of all destination edges.
-        self.edges: _Dict[str, _Set[_GraphEdge]] = {}
+        self.edges: dict[str, set[_GraphEdge]] = {}
 
     def handle_node(self, label: str, node: T = None) -> T:
         """

vcorelib/graph/edge.py

@@ -3,15 +3,14 @@
 """
 
 # built-in
-from typing import Dict as _Dict
 from typing import NamedTuple
 from typing import Optional as _Optional
 from typing import TextIO as _TextIO
 
 # internal
 from vcorelib.graph.port import Port as _Port
 
-AttributeMap = _Dict[str, str]
+AttributeMap = dict[str, str]
 
 
 def write_attributes(stream: _TextIO, data: AttributeMap = None) -> None:

vcorelib/graph/port.py

@@ -7,10 +7,8 @@
 
 # built-in
 from itertools import zip_longest as _zip_longest
-from typing import Dict as _Dict
 from typing import NamedTuple
 from typing import Optional as _Optional
-from typing import Set as _Set
 from typing import TextIO as _TextIO
 
 
@@ -79,22 +77,22 @@ class PortManager:
     def __init__(self) -> None:
         """Initialize this port manager."""
 
-        self.outputs: _Dict[str, Port] = {}
-        self.inputs: _Dict[str, Port] = {}
+        self.outputs: dict[str, Port] = {}
+        self.inputs: dict[str, Port] = {}
 
-    def inout_labels(self) -> _Set[str]:
+    def inout_labels(self) -> set[str]:
         """Get inout port labels."""
         return set(self.inputs.keys()).intersection(self.outputs.keys())
 
-    def input_labels(self, exclude_inout: bool = True) -> _Set[str]:
+    def input_labels(self, exclude_inout: bool = True) -> set[str]:
         """Get input port labels."""
 
         result = set(self.inputs.keys())
         if exclude_inout:
             result -= self.inout_labels()
         return result
 
-    def output_labels(self, exclude_inout: bool = True) -> _Set[str]:
+    def output_labels(self, exclude_inout: bool = True) -> set[str]:
         """Get output port labels."""
 
         result = set(self.outputs.keys())

vcorelib/io/arbiter/base.py

@@ -6,7 +6,6 @@
 from io import StringIO
 import logging
 from pathlib import Path
-from typing import List
 from typing import Optional as _Optional
 
 # third-party
@@ -143,7 +142,7 @@ async def decode_async(
         maxsplit: int = 1,
         expect_overwrite: bool = False,
         strategy: MergeStrategy = MergeStrategy.RECURSIVE,
-        files_loaded: List[Path] = None,
+        files_loaded: list[Path] = None,
         **kwargs,
     ) -> LoadResult:
         """Attempt to load data from a file using asyncio."""
@@ -229,7 +228,7 @@ def decode(
         maxsplit: int = 1,
         expect_overwrite: bool = False,
         strategy: MergeStrategy = MergeStrategy.RECURSIVE,
-        files_loaded: List[Path] = None,
+        files_loaded: list[Path] = None,
         **kwargs,
     ) -> LoadResult:
         """Attempt to load data from a file."""

vcorelib/io/archive/__init__.py

@@ -10,7 +10,6 @@
 import shutil
 import tarfile
 from typing import Optional as _Optional
-from typing import Tuple as _Tuple
 import zipfile
 
 # internal
@@ -58,7 +57,7 @@ def extractall(
     maxsplit: int = 1,
     tar_filter: str = "tar",
     **extract_kwargs,
-) -> _Tuple[bool, int]:
+) -> tuple[bool, int]:
     """
     Attempt to extract an arbitrary archive to a destination. Return whether or
     not this succeeded and how long it took.
@@ -96,7 +95,7 @@ def make_archive(
     ext_str: str = _DEFAULT_ARCHIVE_EXT,
     dst_dir: Path = None,
     **archive_kwargs,
-) -> _Tuple[_Optional[Path], int]:
+) -> tuple[_Optional[Path], int]:
     """
     Create an archive from a source directory, named after that directory,
     and optionally moved to a destination other than the parent directory

vcorelib/io/file_writer.py

@@ -8,7 +8,7 @@
 from io import StringIO
 import os
 from pathlib import Path
-from typing import Callable, Iterator, List, Optional, TextIO, Tuple
+from typing import Callable, Iterator, Optional, TextIO
 
 # third-party
 import markdown
@@ -38,8 +38,8 @@ def wrap(self, data: str) -> str:
         return f"# {data}"
 
 
-LineWithComment = Tuple[str, Optional[str]]
-LinesWithComments = List[LineWithComment]
+LineWithComment = tuple[str, Optional[str]]
+LinesWithComments = list[LineWithComment]
 
 MARKDOWN_EXTENSIONS = ["extra"]
 

vcorelib/io/types.py

@@ -11,7 +11,6 @@
 from typing import NamedTuple
 from typing import Optional as _Optional
 from typing import TextIO
-from typing import Tuple as _Tuple
 from typing import Union as _Union
 
 # third-party
@@ -168,7 +167,7 @@ def __eq__(self, other: object) -> bool:
         assert isinstance(other, (LoadResult, tuple))
         return bool(self.data == other[0] and self.success == other[1])
 
-    def require_success(self, path: _Union[Path, str]) -> None:
+    def require_success(self, path: Path | str) -> None:
         """Raise a canonical exception if this result is a failure."""
         assert self.success, f"Couldn't load '{path}'!"
 
@@ -192,7 +191,7 @@ def merge(
         )
 
 
-EncodeResult = _Tuple[bool, int]
+EncodeResult = tuple[bool, int]
 DataStream = _Union[TextIO, StringIO]
 StreamProcessor = _Callable[[DataStream], DataStream]
 DataDecoder = _Callable[[DataStream, LoggerType], LoadResult]

vcorelib/logging/__init__.py

@@ -10,7 +10,6 @@
 from logging.handlers import QueueHandler
 from queue import SimpleQueue
 from typing import Iterator as _Iterator
-from typing import Tuple
 
 # internal
 from vcorelib.logging.args import (
@@ -60,7 +59,7 @@ def queue_handler(
     queue: LogRecordQueue = None,
     handler: QueueHandler = None,
     root_formatter: bool = True,
-) -> Tuple[LogRecordQueue, QueueHandler]:
+) -> tuple[LogRecordQueue, QueueHandler]:
     """
     Set up and return a simple queue and logging queue handler. Use the
     provided objects if they already exist.