Modernize typing syntax to Python 3.10+ standards

.github/workflows/run_tests.yml

@@ -55,7 +55,7 @@ jobs:
       fail-fast: false
       matrix:
         subset: [backends, slow_tests, group_a, group_b]
-        os: [macos-13, macos-14, ubuntu-22.04] # Operating systems
+        os: [macos-14, ubuntu-22.04] # Operating systems
         compiler: [8] # GNU compiler version
         rai: [1.2.7] # Redis AI versions
         py_v: ["3.10", "3.11", "3.12"] # Python versions

conftest.py

@@ -64,6 +64,7 @@
     RunSettings,
     SrunSettings,
 )
+from collections.abc import Callable, Collection
 
 logger = get_logger(__name__)
 
@@ -79,7 +80,7 @@
 test_alloc_specs_path = os.getenv("SMARTSIM_TEST_ALLOC_SPEC_SHEET_PATH", None)
 test_ports = CONFIG.test_ports
 test_account = CONFIG.test_account or ""
-test_batch_resources: t.Dict[t.Any, t.Any] = CONFIG.test_batch_resources
+test_batch_resources: dict[t.Any, t.Any] = CONFIG.test_batch_resources
 test_output_dirs = 0
 mpi_app_exe = None
 built_mpi_app = False
@@ -169,7 +170,7 @@ def pytest_sessionfinish(
         kill_all_test_spawned_processes()
 
 
-def build_mpi_app() -> t.Optional[pathlib.Path]:
+def build_mpi_app() -> pathlib.Path | None:
     global built_mpi_app
     built_mpi_app = True
     cc = shutil.which("cc")
@@ -190,7 +191,7 @@ def build_mpi_app() -> t.Optional[pathlib.Path]:
         return None
 
 @pytest.fixture(scope="session")
-def mpi_app_path() -> t.Optional[pathlib.Path]:
+def mpi_app_path() -> pathlib.Path | None:
     """Return path to MPI app if it was built
 
         return None if it could not or will not be built
@@ -223,7 +224,7 @@ def kill_all_test_spawned_processes() -> None:
 
 
 
-def get_hostlist() -> t.Optional[t.List[str]]:
+def get_hostlist() -> list[str] | None:
     global test_hostlist
     if not test_hostlist:
         if "PBS_NODEFILE" in os.environ and test_launcher == "pals":
@@ -251,14 +252,14 @@ def get_hostlist() -> t.Optional[t.List[str]]:
     return test_hostlist
 
 
-def _parse_hostlist_file(path: str) -> t.List[str]:
+def _parse_hostlist_file(path: str) -> list[str]:
     with open(path, "r", encoding="utf-8") as nodefile:
         return list({line.strip() for line in nodefile.readlines()})
 
 
 @pytest.fixture(scope="session")
-def alloc_specs() -> t.Dict[str, t.Any]:
-    specs: t.Dict[str, t.Any] = {}
+def alloc_specs() -> dict[str, t.Any]:
+    specs: dict[str, t.Any] = {}
     if test_alloc_specs_path:
         try:
             with open(test_alloc_specs_path, encoding="utf-8") as spec_file:
@@ -293,7 +294,7 @@ def _reset():
 )
 
 
-def _find_free_port(ports: t.Collection[int]) -> int:
+def _find_free_port(ports: Collection[int]) -> int:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         for port in ports:
             try:
@@ -310,7 +311,7 @@ def _find_free_port(ports: t.Collection[int]) -> int:
 
 
 @pytest.fixture(scope="session")
-def wlmutils() -> t.Type[WLMUtils]:
+def wlmutils() -> type[WLMUtils]:
     return WLMUtils
 
 
@@ -335,22 +336,22 @@ def get_test_account() -> str:
         return get_account()
 
     @staticmethod
-    def get_test_interface() -> t.List[str]:
+    def get_test_interface() -> list[str]:
         return test_nic
 
     @staticmethod
-    def get_test_hostlist() -> t.Optional[t.List[str]]:
+    def get_test_hostlist() -> list[str] | None:
         return get_hostlist()
 
     @staticmethod
-    def get_batch_resources() -> t.Dict:
+    def get_batch_resources() -> dict:
         return test_batch_resources
 
     @staticmethod
     def get_base_run_settings(
-        exe: str, args: t.List[str], nodes: int = 1, ntasks: int = 1, **kwargs: t.Any
+        exe: str, args: list[str], nodes: int = 1, ntasks: int = 1, **kwargs: t.Any
     ) -> RunSettings:
-        run_args: t.Dict[str, t.Union[int, str, float, None]] = {}
+        run_args: dict[str, int | str | float | None] = {}
 
         if test_launcher == "slurm":
             run_args = {"--nodes": nodes, "--ntasks": ntasks, "--time": "00:10:00"}
@@ -391,9 +392,9 @@ def get_base_run_settings(
 
     @staticmethod
     def get_run_settings(
-        exe: str, args: t.List[str], nodes: int = 1, ntasks: int = 1, **kwargs: t.Any
+        exe: str, args: list[str], nodes: int = 1, ntasks: int = 1, **kwargs: t.Any
     ) -> RunSettings:
-        run_args: t.Dict[str, t.Union[int, str, float, None]] = {}
+        run_args: dict[str, int | str | float | None] = {}
 
         if test_launcher == "slurm":
             run_args = {"nodes": nodes, "ntasks": ntasks, "time": "00:10:00"}
@@ -423,7 +424,7 @@ def get_run_settings(
         return RunSettings(exe, args)
 
     @staticmethod
-    def choose_host(rs: RunSettings) -> t.Optional[str]:
+    def choose_host(rs: RunSettings) -> str | None:
         if isinstance(rs, (MpirunSettings, MpiexecSettings)):
             hl = get_hostlist()
             if hl is not None:
@@ -450,13 +451,13 @@ def check_output_dir() -> None:
 
 
 @pytest.fixture
-def dbutils() -> t.Type[DBUtils]:
+def dbutils() -> type[DBUtils]:
     return DBUtils
 
 
 class DBUtils:
     @staticmethod
-    def get_db_configs() -> t.Dict[str, t.Any]:
+    def get_db_configs() -> dict[str, t.Any]:
         config_settings = {
             "enable_checkpoints": 1,
             "set_max_memory": "3gb",
@@ -470,7 +471,7 @@ def get_db_configs() -> t.Dict[str, t.Any]:
         return config_settings
 
     @staticmethod
-    def get_smartsim_error_db_configs() -> t.Dict[str, t.Any]:
+    def get_smartsim_error_db_configs() -> dict[str, t.Any]:
         bad_configs = {
             "save": [
                 "-1",  # frequency must be positive
@@ -497,8 +498,8 @@ def get_smartsim_error_db_configs() -> t.Dict[str, t.Any]:
         return bad_configs
 
     @staticmethod
-    def get_type_error_db_configs() -> t.Dict[t.Union[int, str], t.Any]:
-        bad_configs: t.Dict[t.Union[int, str], t.Any] = {
+    def get_type_error_db_configs() -> dict[int | str, t.Any]:
+        bad_configs: dict[int | str, t.Any] = {
             "save": [2, True, ["2"]],  # frequency must be specified as a string
             "maxmemory": [99, True, ["99"]],  # memory form must be a string
             "maxclients": [3, True, ["3"]],  # number of clients must be a string
@@ -519,9 +520,9 @@ def get_type_error_db_configs() -> t.Dict[t.Union[int, str], t.Any]:
     @staticmethod
     def get_config_edit_method(
         db: Orchestrator, config_setting: str
-    ) -> t.Optional[t.Callable[..., None]]:
+    ) -> Callable[..., None] | None:
         """Get a db configuration file edit method from a str"""
-        config_edit_methods: t.Dict[str, t.Callable[..., None]] = {
+        config_edit_methods: dict[str, Callable[..., None]] = {
             "enable_checkpoints": db.enable_checkpoints,
             "set_max_memory": db.set_max_memory,
             "set_eviction_strategy": db.set_eviction_strategy,
@@ -564,7 +565,7 @@ def test_dir(request: pytest.FixtureRequest) -> str:
 
 
 @pytest.fixture
-def fileutils() -> t.Type[FileUtils]:
+def fileutils() -> type[FileUtils]:
     return FileUtils
 
 
@@ -589,7 +590,7 @@ def get_test_dir_path(dirname: str) -> str:
 
     @staticmethod
     def make_test_file(
-        file_name: str, file_dir: str, file_content: t.Optional[str] = None
+        file_name: str, file_dir: str, file_content: str | None = None
     ) -> str:
         """Create a dummy file in the test output directory.
 
@@ -609,7 +610,7 @@ def make_test_file(
 
 
 @pytest.fixture
-def mlutils() -> t.Type[MLUtils]:
+def mlutils() -> type[MLUtils]:
     return MLUtils
 
 
@@ -624,21 +625,21 @@ def get_test_num_gpus() -> int:
 
 
 @pytest.fixture
-def coloutils() -> t.Type[ColoUtils]:
+def coloutils() -> type[ColoUtils]:
     return ColoUtils
 
 
 class ColoUtils:
     @staticmethod
     def setup_test_colo(
-        fileutils: t.Type[FileUtils],
+        fileutils: type[FileUtils],
         db_type: str,
         exp: Experiment,
         application_file: str,
-        db_args: t.Dict[str, t.Any],
-        colo_settings: t.Optional[RunSettings] = None,
+        db_args: dict[str, t.Any],
+        colo_settings: RunSettings | None = None,
         colo_model_name: str = "colocated_model",
-        port: t.Optional[int] = None,
+        port: int | None = None,
         on_wlm: bool = False,
     ) -> Model:
         """Setup database needed for the colo pinning tests"""
@@ -666,7 +667,7 @@ def setup_test_colo(
             socket_name = f"{colo_model_name}_{socket_suffix}.socket"
             db_args["unix_socket"] = os.path.join(tmp_dir, socket_name)
 
-        colocate_fun: t.Dict[str, t.Callable[..., None]] = {
+        colocate_fun: dict[str, Callable[..., None]] = {
             "tcp": colo_model.colocate_db_tcp,
             "deprecated": colo_model.colocate_db,
             "uds": colo_model.colocate_db_uds,
@@ -708,7 +709,7 @@ def config() -> Config:
 class CountingCallable:
     def __init__(self) -> None:
         self._num: int = 0
-        self._details: t.List[t.Tuple[t.Tuple[t.Any, ...], t.Dict[str, t.Any]]] = []
+        self._details: list[tuple[tuple[t.Any, ...], dict[str, t.Any]]] = []
 
     def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
         self._num += 1
@@ -719,12 +720,12 @@ def num_calls(self) -> int:
         return self._num
 
     @property
-    def details(self) -> t.List[t.Tuple[t.Tuple[t.Any, ...], t.Dict[str, t.Any]]]:
+    def details(self) -> list[tuple[tuple[t.Any, ...], dict[str, t.Any]]]:
         return self._details
 
 ## Reuse database across tests
 
-database_registry: t.DefaultDict[str, t.Optional[Orchestrator]] = defaultdict(lambda: None)
+database_registry: defaultdict[str, Orchestrator | None] = defaultdict(lambda: None)
 
 @pytest.fixture(scope="function")
 def local_experiment(test_dir: str) -> smartsim.Experiment:
@@ -758,13 +759,13 @@ class DBConfiguration:
     name: str
     launcher: str
     num_nodes: int
-    interface: t.Union[str,t.List[str]]
-    hostlist: t.Optional[t.List[str]]
+    interface: str | list[str]
+    hostlist: list[str] | None
     port: int
 
 @dataclass
 class PrepareDatabaseOutput:
-    orchestrator: t.Optional[Orchestrator] # The actual orchestrator object
+    orchestrator: Orchestrator | None # The actual orchestrator object
     new_db: bool     # True if a new database was created when calling prepare_db
 
 # Reuse databases
@@ -817,7 +818,7 @@ def clustered_db(wlmutils: WLMUtils) -> t.Generator[DBConfiguration, None, None]
 
 
 @pytest.fixture
-def register_new_db() -> t.Callable[[DBConfiguration], Orchestrator]:
+def register_new_db() -> Callable[[DBConfiguration], Orchestrator]:
     def _register_new_db(
         config: DBConfiguration
     ) -> Orchestrator:
@@ -845,11 +846,11 @@ def _register_new_db(
 
 @pytest.fixture(scope="function")
 def prepare_db(
-    register_new_db: t.Callable[
+    register_new_db: Callable[
         [DBConfiguration],
         Orchestrator
     ]
-) -> t.Callable[
+) -> Callable[
     [DBConfiguration],
     PrepareDatabaseOutput
 ]:

doc/changelog.md

@@ -11,6 +11,7 @@ To be released at some point in the future
 
 Description
 
+- Modernize typing syntax to Python 3.10+ standards
 - Removed telemetry functionality, LaunchedManifest tracking
   classes, and SmartDashboard integration
 - Update copyright headers from 2021-2024 to 2021-2025 across the entire codebase
@@ -24,6 +25,10 @@ Description
 
 Detailed Notes
 
+- Modernized typing syntax to use Python 3.10+ standards, replacing
+  `Union[X, Y]` with `X | Y`, `Optional[X]` with `X | None`, and generic
+  collections (`List[X]` → `list[X]`, `Dict[X, Y]` → `dict[X, Y]`, etc.).
+  ([SmartSim-PR791](https://github.com/CrayLabs/SmartSim/pull/791))
 - Removed telemetry functionality, LaunchedManifest tracking
   system, and SmartDashboard integration.
   This includes complete removal of the telemetry monitor and collection system,

smartsim/_core/_cli/build.py

@@ -31,7 +31,7 @@
 import re
 import shutil
 import textwrap
-import typing as t
+from collections.abc import Callable, Collection
 from pathlib import Path
 
 from tabulate import tabulate
@@ -139,7 +139,7 @@ def build_redis_ai(
 
 def parse_requirement(
     requirement: str,
-) -> t.Tuple[str, t.Optional[str], t.Callable[[Version_], bool]]:
+) -> tuple[str, str | None, Callable[[Version_], bool]]:
     operators = {
         "==": operator.eq,
         "<=": operator.le,
@@ -199,10 +199,10 @@ def check_ml_python_packages(packages: MLPackageCollection) -> None:
 
 
 def _format_incompatible_python_env_message(
-    missing: t.Collection[str], conflicting: t.Collection[str]
+    missing: Collection[str], conflicting: Collection[str]
 ) -> str:
     indent = "\n\t"
-    fmt_list: t.Callable[[str, t.Collection[str]], str] = lambda n, l: (
+    fmt_list: Callable[[str, Collection[str]], str] = lambda n, l: (
         f"{n}:{indent}{indent.join(l)}" if l else ""
     )
     missing_str = fmt_list("Missing", missing)
@@ -237,7 +237,7 @@ def _configure_keydb_build(versions: Versioner) -> None:
 
 # pylint: disable-next=too-many-statements
 def execute(
-    args: argparse.Namespace, _unparsed_args: t.Optional[t.List[str]] = None, /
+    args: argparse.Namespace, _unparsed_args: list[str] | None = None, /
 ) -> int:
 
     # Unpack various arguments

smartsim/_core/_cli/clean.py

@@ -25,7 +25,6 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 import argparse
-import typing as t
 
 from smartsim._core._cli.utils import clean, get_install_path
 
@@ -41,13 +40,13 @@ def configure_parser(parser: argparse.ArgumentParser) -> None:
 
 
 def execute(
-    args: argparse.Namespace, _unparsed_args: t.Optional[t.List[str]] = None, /
+    args: argparse.Namespace, _unparsed_args: list[str] | None = None, /
 ) -> int:
     return clean(get_install_path() / "_core", _all=args.clobber)
 
 
 def execute_all(
-    args: argparse.Namespace, _unparsed_args: t.Optional[t.List[str]] = None, /
+    args: argparse.Namespace, _unparsed_args: list[str] | None = None, /
 ) -> int:
     args.clobber = True
     return execute(args)