daemon: updated top level core modules from using deprecated type hinting

daemon/core/utils.py

@@ -17,23 +17,11 @@ import shutil
 import sys
 import threading
 from collections import OrderedDict
+from collections.abc import Iterable
 from pathlib import Path
 from queue import Queue
 from subprocess import PIPE, STDOUT, Popen
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Callable,
-    Dict,
-    Generic,
-    Iterable,
-    List,
-    Optional,
-    Tuple,
-    Type,
-    TypeVar,
-    Union,
-)
+from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union
 
 import netaddr
 
@@ -70,7 +58,7 @@ def execute_script(coreemu: "CoreEmu", file_path: Path, args: str) -> None:
 
 
 def execute_file(
-    path: Path, exec_globals: Dict[str, str] = None, exec_locals: Dict[str, str] = None
+    path: Path, exec_globals: dict[str, str] = None, exec_locals: dict[str, str] = None
 ) -> None:
     """
     Provides a way to execute a file.
@@ -131,7 +119,7 @@ def _valid_module(path: Path) -> bool:
     return True
 
 
-def _is_class(module: Any, member: Type, clazz: Type) -> bool:
+def _is_class(module: Any, member: type, clazz: type) -> bool:
     """
     Validates if a module member is a class and an instance of a CoreService.
 
@@ -175,7 +163,7 @@ def which(command: str, required: bool) -> str:
     return found_path
 
 
-def make_tuple_fromstr(s: str, value_type: Callable[[str], T]) -> Tuple[T]:
+def make_tuple_fromstr(s: str, value_type: Callable[[str], T]) -> tuple[T]:
     """
     Create a tuple from a string.
 
@@ -193,7 +181,7 @@ def make_tuple_fromstr(s: str, value_type: Callable[[str], T]) -> Tuple[T]:
     return tuple(value_type(i) for i in values)
 
 
-def mute_detach(args: str, **kwargs: Dict[str, Any]) -> int:
+def mute_detach(args: str, **kwargs: dict[str, Any]) -> int:
     """
     Run a muted detached process by forking it.
 
@@ -210,7 +198,7 @@ def mute_detach(args: str, **kwargs: Dict[str, Any]) -> int:
 
 def cmd(
     args: str,
-    env: Dict[str, str] = None,
+    env: dict[str, str] = None,
     cwd: Path = None,
     wait: bool = True,
     shell: bool = False,
@@ -249,7 +237,7 @@ def cmd(
         raise CoreCommandError(1, input_args, "", e.strerror)
 
 
-def run_cmds(args: List[str], wait: bool = True, shell: bool = False) -> List[str]:
+def run_cmds(args: list[str], wait: bool = True, shell: bool = False) -> list[str]:
     """
     Execute a series of commands on the host and returns a list of the combined stderr
     stdout output.
@@ -294,7 +282,7 @@ def file_demunge(pathname: str, header: str) -> None:
     :param header: header text to target for removal
     :return: nothing
     """
-    with open(pathname, "r") as read_file:
+    with open(pathname) as read_file:
         lines = read_file.readlines()
 
     start = None
@@ -348,7 +336,7 @@ def sysctl_devname(devname: str) -> Optional[str]:
     return devname.replace(".", "/")
 
 
-def load_config(file_path: Path, d: Dict[str, str]) -> None:
+def load_config(file_path: Path, d: dict[str, str]) -> None:
     """
     Read key=value pairs from a file, into a dict. Skip comments; strip newline
     characters and spacing.
@@ -369,7 +357,7 @@ def load_config(file_path: Path, d: Dict[str, str]) -> None:
             logger.exception("error reading file to dict: %s", file_path)
 
 
-def load_module(import_statement: str, clazz: Generic[T]) -> List[T]:
+def load_module(import_statement: str, clazz: Generic[T]) -> list[T]:
     classes = []
     try:
         module = importlib.import_module(import_statement)
@@ -384,7 +372,7 @@ def load_module(import_statement: str, clazz: Generic[T]) -> List[T]:
     return classes
 
 
-def load_classes(path: Path, clazz: Generic[T]) -> List[T]:
+def load_classes(path: Path, clazz: Generic[T]) -> list[T]:
     """
     Dynamically load classes for use within CORE.
 
@@ -426,12 +414,12 @@ def load_logging_config(config_path: Path) -> None:
 
 
 def run_cmds_threaded(
-    nodes: List["CoreNode"],
-    cmds: List[str],
+    nodes: list["CoreNode"],
+    cmds: list[str],
     wait: bool = True,
     shell: bool = False,
     workers: int = None,
-) -> Tuple[Dict[int, List[str]], List[Exception]]:
+) -> tuple[dict[int, list[str]], list[Exception]]:
     """
     Run a set of commands in order across a provided set of nodes. Each node will
     run the commands within the context of a threadpool.
@@ -446,8 +434,8 @@ def run_cmds_threaded(
     """
 
     def _node_cmds(
-        _target: "CoreNode", _cmds: List[str], _wait: bool, _shell: bool
-    ) -> List[str]:
+        _target: "CoreNode", _cmds: list[str], _wait: bool, _shell: bool
+    ) -> list[str]:
         outputs = []
         for _cmd in _cmds:
             output = _target.cmd(_cmd, wait=_wait, shell=_shell)
@@ -475,12 +463,12 @@ def run_cmds_threaded(
 
 
 def run_cmds_mp(
-    nodes: List["CoreNode"],
-    cmds: List[str],
+    nodes: list["CoreNode"],
+    cmds: list[str],
     wait: bool = True,
     shell: bool = False,
     workers: int = None,
-) -> Tuple[Dict[int, List[str]], List[Exception]]:
+) -> tuple[dict[int, list[str]], list[Exception]]:
     """
     Run a set of commands in order across a provided set of nodes. Each node will
     run the commands within the context of a process pool. This will not work
@@ -521,8 +509,8 @@ def run_cmds_mp(
 
 
 def threadpool(
-    funcs: List[Tuple[Callable, Iterable[Any], Dict[Any, Any]]], workers: int = 10
-) -> Tuple[List[Any], List[Exception]]:
+    funcs: list[tuple[Callable, Iterable[Any], dict[Any, Any]]], workers: int = 10
+) -> tuple[list[Any], list[Exception]]:
     """
     Run provided functions, arguments, and keywords within a threadpool
     collecting results and exceptions.
@@ -575,7 +563,7 @@ def iface_config_id(node_id: int, iface_id: int = None) -> int:
         return node_id
 
 
-def parse_iface_config_id(config_id: int) -> Tuple[int, Optional[int]]:
+def parse_iface_config_id(config_id: int) -> tuple[int, Optional[int]]:
     """
     Parses configuration id, that may be potentially derived from an interface for a
     node.