Leverage PEP-585

docs/source/conf.py

@@ -11,14 +11,14 @@
 #
 # All configuration values have a default; values that are commented out
 # serve to show the default.
+from __future__ import annotations
 
 import importlib
 import os
 import re
 import sys
 from inspect import getmembers, isclass, isfunction
 from pathlib import Path
-from typing import List, Tuple
 
 from click import secho, style
 
@@ -55,7 +55,7 @@
     "notfound.extension",
 ]
 
-# enable autosummary plugin (table of contents for modules/classes/class
+# enable autosummary plugin  (table of contents for modules/classes/class
 # methods)
 autosummary_generate = True
 autosummary_generate_overwrite = False
@@ -102,6 +102,9 @@
         "tuple",
         "Any",
         "Dict",
+        "dict",
+        "list",
+        "set",
         "typing.Dict",
         "typing.Iterable",
         "typing.List",
@@ -122,7 +125,9 @@
         "kedro.framework.context.context.KedroContext",
         "kedro.framework.startup.ProjectMetadata",
         "abc.ABC",
+        "Path",
         "pathlib.Path",
+        "PurePosixPath",
         "pathlib.PurePosixPath",
         "requests.auth.AuthBase",
         "google.oauth2.credentials.Credentials",
@@ -133,6 +138,7 @@
         "kedro.extras.datasets.pandas.json_dataset.JSONDataSet",
         "kedro_datasets.pandas.json_dataset.JSONDataSet",
         "pluggy._manager.PluginManager",
+        "PluginManager",
         "_DI",
         "_DO",
         # The statements below were added after subclassing UserDict in AbstractConfigLoader.
@@ -330,7 +336,7 @@ def remove_arrows_in_examples(lines):
         lines[i] = line.replace(">>>", "")
 
 
-def autolink_replacements(what: str) -> List[Tuple[str, str, str]]:
+def autolink_replacements(what: str) -> list[tuple[str, str, str]]:
     """
     Create a list containing replacement tuples of the form:
     (``regex``, ``replacement``, ``obj``) for all classes and methods which are
@@ -402,7 +408,7 @@ def autolink_replacements(what: str) -> List[Tuple[str, str, str]]:
     return replacements, suggestions
 
 
-def log_suggestions(lines: List[str], name: str):
+def log_suggestions(lines: list[str], name: str):
     """Use the ``suggestions`` list to log in the terminal places where the
     developer has forgotten to surround with double back-ticks class
     name/function name references.

features/environment.py

@@ -1,16 +1,16 @@
 """Behave environment setup commands."""
 # pylint: disable=unused-argument
+from __future__ import annotations
 
 import os
 import shutil
 import tempfile
 import venv
 from pathlib import Path
-from typing import Set
 
 from features.steps.sh_run import run
 
-_PATHS_TO_REMOVE: Set[Path] = set()
+_PATHS_TO_REMOVE: set[Path] = set()
 
 FRESH_VENV_TAG = "fresh_venv"
 

features/steps/sh_run.py

@@ -1,12 +1,14 @@
+from __future__ import annotations
+
 import shlex
 import subprocess
-from typing import Any, List, Union
+from typing import Any
 
 import psutil
 
 
 def run(
-    cmd: Union[list, str], split: bool = True, print_output: bool = False, **kwargs: Any
+    cmd: list | str, split: bool = True, print_output: bool = False, **kwargs: Any
 ) -> subprocess.CompletedProcess:
     """Run a shell command.
 
@@ -45,7 +47,7 @@ def run(
     return result
 
 
-def check_run(cmd: Union[list, str], print_output: bool = False) -> None:
+def check_run(cmd: list | str, print_output: bool = False) -> None:
     """
     Run cmd using subprocess.check_call (throws error if non-zero value
     returned)
@@ -74,7 +76,7 @@ class ChildTerminatingPopen(subprocess.Popen):
         dies (so-called orphan processes)
     """
 
-    def __init__(self, cmd: List[str], **kwargs) -> None:
+    def __init__(self, cmd: list[str], **kwargs) -> None:
         """
         Initializer pipes stderr and stdout.
 

features/steps/test_starter/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/pipeline_registry.py

@@ -1,11 +1,11 @@
 """Project pipelines."""
-from typing import Dict
+from __future__ import annotations
 
 from kedro.framework.project import find_pipelines
 from kedro.pipeline import Pipeline, pipeline
 
 
-def register_pipelines() -> Dict[str, Pipeline]:
+def register_pipelines() -> dict[str, Pipeline]:
     """Register the project's pipelines.
 
     Returns:

features/steps/test_starter/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/pipelines/data_engineering/nodes.py

@@ -3,13 +3,14 @@
 
 PLEASE DELETE THIS FILE ONCE YOU START WORKING ON YOUR OWN PROJECT!
 """
+from __future__ import annotations
 
-from typing import Any, Dict
+from typing import Any
 
 import pandas as pd
 
 
-def split_data(data: pd.DataFrame, example_test_data_ratio: float) -> Dict[str, Any]:
+def split_data(data: pd.DataFrame, example_test_data_ratio: float) -> dict[str, Any]:
     """Node for splitting the classical Iris data set into training and test
     sets, each split into features and labels.
     The split ratio parameter is taken from conf/project/parameters.yml.

features/steps/test_starter/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/pipelines/data_science/nodes.py

@@ -4,16 +4,17 @@
 Delete this when you start working on your own Kedro project.
 """
 # pylint: disable=invalid-name
+from __future__ import annotations
 
 import logging
-from typing import Any, Dict
+from typing import Any
 
 import numpy as np
 import pandas as pd
 
 
 def train_model(
-    train_x: pd.DataFrame, train_y: pd.DataFrame, parameters: Dict[str, Any]
+    train_x: pd.DataFrame, train_y: pd.DataFrame, parameters: dict[str, Any]
 ) -> np.ndarray:
     """Node for training a simple multi-class logistic regression model. The
     number of training iterations as well as the learning rate are taken from

features/steps/util.py

@@ -1,12 +1,13 @@
 """Common functions for e2e testing.
 """
+from __future__ import annotations
 
 import os
 import re
 from contextlib import contextmanager
 from pathlib import Path
 from time import sleep, time
-from typing import Any, Callable, Iterator, List
+from typing import Any, Callable, Iterator
 
 
 @contextmanager
@@ -72,7 +73,7 @@ def wait_for(
     )
 
 
-def parse_csv(text: str) -> List[str]:
+def parse_csv(text: str) -> list[str]:
     """Parse comma separated **double quoted** strings in behave steps
 
     Args:

kedro/config/abstract_config.py

@@ -1,8 +1,10 @@
 """This module provides ``kedro.abstract_config`` with the baseline
 class model for a `ConfigLoader` implementation.
 """
+from __future__ import annotations
+
 from collections import UserDict
-from typing import Any, Dict
+from typing import Any
 
 
 class AbstractConfigLoader(UserDict):
@@ -16,8 +18,8 @@ def __init__(
         self,
         conf_source: str,
         env: str = None,
-        runtime_params: Dict[str, Any] = None,
-        **kwargs
+        runtime_params: dict[str, Any] = None,
+        **kwargs,
     ):
         super().__init__()
         self.conf_source = conf_source

kedro/config/common.py

@@ -1,11 +1,12 @@
 """This module contains methods and facade interfaces for various ConfigLoader
 implementations.
 """
+from __future__ import annotations
 
 import logging
 from glob import iglob
 from pathlib import Path
-from typing import AbstractSet, Any, Dict, Iterable, List, Set
+from typing import AbstractSet, Any, Iterable
 from warnings import warn
 
 from yaml.parser import ParserError
@@ -28,8 +29,8 @@ def _get_config_from_patterns(
     conf_paths: Iterable[str],
     patterns: Iterable[str] = None,
     ac_template: bool = False,
-    ac_context: Dict[str, Any] = None,
-) -> Dict[str, Any]:
+    ac_context: dict[str, Any] = None,
+) -> dict[str, Any]:
     """Recursively scan for configuration files, load and merge them, and
     return them in the form of a config dictionary.
 
@@ -64,8 +65,8 @@ def _get_config_from_patterns(
             "pattern to match config filenames against."
         )
 
-    config: Dict[str, Any] = {}
-    processed_files: Set[Path] = set()
+    config: dict[str, Any] = {}
+    processed_files: set[Path] = set()
 
     for conf_path in conf_paths:
         if not Path(conf_path).is_dir():
@@ -104,8 +105,8 @@ def _get_config_from_patterns(
 
 
 def _load_config_file(
-    config_file: Path, ac_template: bool = False, ac_context: Dict[str, Any] = None
-) -> Dict[str, Any]:
+    config_file: Path, ac_template: bool = False, ac_context: dict[str, Any] = None
+) -> dict[str, Any]:
     """Load an individual config file using `anyconfig` as a backend.
 
     Args:
@@ -149,8 +150,8 @@ def _load_config_file(
 
 
 def _load_configs(
-    config_filepaths: List[Path], ac_template: bool, ac_context: Dict[str, Any] = None
-) -> Dict[str, Any]:
+    config_filepaths: list[Path], ac_template: bool, ac_context: dict[str, Any] = None
+) -> dict[str, Any]:
     """Recursively load all configuration files, which satisfy
     a given list of glob patterns from a specific path.
 
@@ -173,7 +174,7 @@ def _load_configs(
     """
 
     aggregate_config = {}
-    seen_file_to_keys: Dict[Path, AbstractSet[str]] = {}
+    seen_file_to_keys: dict[Path, AbstractSet[str]] = {}
 
     for config_filepath in config_filepaths:
         single_config = _load_config_file(
@@ -189,9 +190,9 @@ def _load_configs(
 def _lookup_config_filepaths(
     conf_path: Path,
     patterns: Iterable[str],
-    processed_files: Set[Path],
+    processed_files: set[Path],
     logger: Any,
-) -> List[Path]:
+) -> list[Path]:
     config_files = _path_lookup(conf_path, patterns)
 
     seen_files = config_files & processed_files
@@ -207,7 +208,7 @@ def _lookup_config_filepaths(
 
 def _remove_duplicates(items: Iterable[str]):
     """Remove duplicates while preserving the order."""
-    unique_items: List[str] = []
+    unique_items: list[str] = []
     for item in items:
         if item not in unique_items:
             unique_items.append(item)
@@ -220,7 +221,7 @@ def _remove_duplicates(items: Iterable[str]):
 
 
 def _check_duplicate_keys(
-    processed_files: Dict[Path, AbstractSet[str]], filepath: Path, conf: Dict[str, Any]
+    processed_files: dict[Path, AbstractSet[str]], filepath: Path, conf: dict[str, Any]
 ) -> None:
     duplicates = []
 
@@ -238,7 +239,7 @@ def _check_duplicate_keys(
         raise ValueError(f"Duplicate keys found in {filepath} and:\n- {dup_str}")
 
 
-def _path_lookup(conf_path: Path, patterns: Iterable[str]) -> Set[Path]:
+def _path_lookup(conf_path: Path, patterns: Iterable[str]) -> set[Path]:
     """Return a set of all configuration files from ``conf_path`` or
     its subdirectories, which satisfy a given list of glob patterns.
 

kedro/config/config.py

@@ -1,8 +1,10 @@
 """This module provides ``kedro.config`` with the functionality to load one
 or more configuration files from specified paths.
 """
+from __future__ import annotations
+
 from pathlib import Path
-from typing import Any, Dict, Iterable, List
+from typing import Any, Iterable
 
 from kedro.config.abstract_config import AbstractConfigLoader
 from kedro.config.common import _get_config_from_patterns, _remove_duplicates
@@ -68,8 +70,8 @@ def __init__(
         self,
         conf_source: str,
         env: str = None,
-        runtime_params: Dict[str, Any] = None,
-        config_patterns: Dict[str, List[str]] = None,
+        runtime_params: dict[str, Any] = None,
+        config_patterns: dict[str, list[str]] = None,
         *,
         base_env: str = "base",
         default_run_env: str = "local",
@@ -125,7 +127,7 @@ def conf_paths(self):
         """Property method to return deduplicated configuration paths."""
         return _remove_duplicates(self._build_conf_paths())
 
-    def get(self, *patterns: str) -> Dict[str, Any]:  # type: ignore
+    def get(self, *patterns: str) -> dict[str, Any]:  # type: ignore
         return _get_config_from_patterns(
             conf_paths=self.conf_paths, patterns=list(patterns)
         )