style(datasets): use PEP 585

kedro-datasets/kedro_datasets/api/api_dataset.py

@@ -1,7 +1,9 @@
 """``APIDataSet`` loads the data from HTTP(S) APIs.
 It uses the python requests library: https://requests.readthedocs.io/en/latest/
 """
-from typing import Any, Dict, List, NoReturn, Tuple, Union
+from __future__ import annotations
+
+from typing import Any, NoReturn, Union
 
 import requests
 from kedro.io.core import AbstractDataSet, DataSetError
@@ -63,8 +65,8 @@ def __init__(
         self,
         url: str,
         method: str = "GET",
-        load_args: Dict[str, Any] = None,
-        credentials: Union[Tuple[str, str], List[str], AuthBase] = None,
+        load_args: dict[str, Any] = None,
+        credentials: Union[tuple[str, str], list[str], AuthBase] = None,
     ) -> None:
         """Creates a new instance of ``APIDataSet`` to fetch data from an API endpoint.
 
@@ -95,7 +97,7 @@ def __init__(
         if "timeout" in self._load_args:
             self._load_args["timeout"] = self._convert_type(self._load_args["timeout"])
 
-        self._request_args: Dict[str, Any] = {
+        self._request_args: dict[str, Any] = {
             "url": url,
             "method": method,
             "auth": self._convert_type(self._auth),
@@ -109,11 +111,11 @@ def _convert_type(value: Any):
         However, for some parameters in the Python requests library,
         only Tuples are allowed.
         """
-        if isinstance(value, List):
+        if isinstance(value, list):
             return tuple(value)
         return value
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         # prevent auth from logging
         request_args_cp = self._request_args.copy()
         request_args_cp.pop("auth", None)

kedro-datasets/kedro_datasets/biosequence/biosequence_dataset.py

@@ -1,16 +1,18 @@
 """BioSequenceDataSet loads and saves data to/from bio-sequence objects to
 file.
 """
+from __future__ import annotations
+
 from copy import deepcopy
 from pathlib import PurePosixPath
-from typing import Any, Dict, List
+from typing import Any
 
 import fsspec
 from Bio import SeqIO
 from kedro.io.core import AbstractDataSet, get_filepath_str, get_protocol_and_path
 
 
-class BioSequenceDataSet(AbstractDataSet[List, List]):
+class BioSequenceDataSet(AbstractDataSet[list, list]):
     r"""``BioSequenceDataSet`` loads and saves data to a sequence file.
 
     Example:
@@ -36,17 +38,17 @@ class BioSequenceDataSet(AbstractDataSet[List, List]):
 
     """
 
-    DEFAULT_LOAD_ARGS: Dict[str, Any] = {}
-    DEFAULT_SAVE_ARGS: Dict[str, Any] = {}
+    DEFAULT_LOAD_ARGS: dict[str, Any] = {}
+    DEFAULT_SAVE_ARGS: dict[str, Any] = {}
 
     # pylint: disable=too-many-arguments
     def __init__(
         self,
         filepath: str,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
-        credentials: Dict[str, Any] = None,
-        fs_args: Dict[str, Any] = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
+        credentials: dict[str, Any] = None,
+        fs_args: dict[str, Any] = None,
     ) -> None:
         """
         Creates a new instance of ``BioSequenceDataSet`` pointing
@@ -100,20 +102,20 @@ def __init__(
         self._fs_open_args_load = _fs_open_args_load
         self._fs_open_args_save = _fs_open_args_save
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         return {
             "filepath": self._filepath,
             "protocol": self._protocol,
             "load_args": self._load_args,
             "save_args": self._save_args,
         }
 
-    def _load(self) -> List:
+    def _load(self) -> list:
         load_path = get_filepath_str(self._filepath, self._protocol)
         with self._fs.open(load_path, **self._fs_open_args_load) as fs_file:
             return list(SeqIO.parse(handle=fs_file, **self._load_args))
 
-    def _save(self, data: List) -> None:
+    def _save(self, data: list) -> None:
         save_path = get_filepath_str(self._filepath, self._protocol)
 
         with self._fs.open(save_path, **self._fs_open_args_save) as fs_file:

kedro-datasets/kedro_datasets/dask/parquet_dataset.py

@@ -1,8 +1,9 @@
 """``ParquetDataSet`` is a data set used to load and save data to parquet files using Dask
 dataframe"""
+from __future__ import annotations
 
 from copy import deepcopy
-from typing import Any, Dict
+from typing import Any
 
 import dask.dataframe as dd
 import fsspec
@@ -83,17 +84,17 @@ class ParquetDataSet(AbstractDataSet[dd.DataFrame, dd.DataFrame]):
               col3: [[int32]]
     """
 
-    DEFAULT_LOAD_ARGS: Dict[str, Any] = {}
-    DEFAULT_SAVE_ARGS: Dict[str, Any] = {"write_index": False}
+    DEFAULT_LOAD_ARGS: dict[str, Any] = {}
+    DEFAULT_SAVE_ARGS: dict[str, Any] = {"write_index": False}
 
     # pylint: disable=too-many-arguments
     def __init__(
         self,
         filepath: str,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
-        credentials: Dict[str, Any] = None,
-        fs_args: Dict[str, Any] = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
+        credentials: dict[str, Any] = None,
+        fs_args: dict[str, Any] = None,
     ) -> None:
         """Creates a new instance of ``ParquetDataSet`` pointing to concrete
         parquet files.
@@ -123,7 +124,7 @@ def __init__(
             self._save_args.update(save_args)
 
     @property
-    def fs_args(self) -> Dict[str, Any]:
+    def fs_args(self) -> dict[str, Any]:
         """Property of optional file system parameters.
 
         Returns:
@@ -133,7 +134,7 @@ def fs_args(self) -> Dict[str, Any]:
         fs_args.update(self._credentials)
         return fs_args
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         return {
             "filepath": self._filepath,
             "load_args": self._load_args,

kedro-datasets/kedro_datasets/email/message_dataset.py

@@ -2,13 +2,15 @@
 using an underlying filesystem (e.g.: local, S3, GCS). It uses the
 ``email`` package in the standard library to manage email messages.
 """
+from __future__ import annotations
+
 from copy import deepcopy
 from email.generator import Generator
 from email.message import Message
 from email.parser import Parser
 from email.policy import default
 from pathlib import PurePosixPath
-from typing import Any, Dict
+from typing import Any
 
 import fsspec
 from kedro.io.core import (
@@ -52,18 +54,18 @@ class EmailMessageDataSet(
 
     """
 
-    DEFAULT_LOAD_ARGS: Dict[str, Any] = {}
-    DEFAULT_SAVE_ARGS: Dict[str, Any] = {}
+    DEFAULT_LOAD_ARGS: dict[str, Any] = {}
+    DEFAULT_SAVE_ARGS: dict[str, Any] = {}
 
     # pylint: disable=too-many-arguments
     def __init__(
         self,
         filepath: str,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
         version: Version = None,
-        credentials: Dict[str, Any] = None,
-        fs_args: Dict[str, Any] = None,
+        credentials: dict[str, Any] = None,
+        fs_args: dict[str, Any] = None,
     ) -> None:
         """Creates a new instance of ``EmailMessageDataSet`` pointing to a concrete text file
         on a specific filesystem.
@@ -140,7 +142,7 @@ def __init__(
         self._fs_open_args_load = _fs_open_args_load
         self._fs_open_args_save = _fs_open_args_save
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         return {
             "filepath": self._filepath,
             "protocol": self._protocol,

kedro-datasets/kedro_datasets/geopandas/geojson_dataset.py

@@ -2,9 +2,11 @@
 underlying functionality is supported by geopandas, so it supports all
 allowed geopandas (pandas) options for loading and saving geosjon files.
 """
+from __future__ import annotations
+
 import copy
 from pathlib import PurePosixPath
-from typing import Any, Dict, Union
+from typing import Any, Union
 
 import fsspec
 import geopandas as gpd
@@ -19,7 +21,7 @@
 
 class GeoJSONDataSet(
     AbstractVersionedDataSet[
-        gpd.GeoDataFrame, Union[gpd.GeoDataFrame, Dict[str, gpd.GeoDataFrame]]
+        gpd.GeoDataFrame, Union[gpd.GeoDataFrame, dict[str, gpd.GeoDataFrame]]
     ]
 ):
     """``GeoJSONDataSet`` loads/saves data to a GeoJSON file using an underlying filesystem
@@ -44,18 +46,18 @@ class GeoJSONDataSet(
 
     """
 
-    DEFAULT_LOAD_ARGS: Dict[str, Any] = {}
+    DEFAULT_LOAD_ARGS: dict[str, Any] = {}
     DEFAULT_SAVE_ARGS = {"driver": "GeoJSON"}
 
     # pylint: disable=too-many-arguments
     def __init__(
         self,
         filepath: str,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
         version: Version = None,
-        credentials: Dict[str, Any] = None,
-        fs_args: Dict[str, Any] = None,
+        credentials: dict[str, Any] = None,
+        fs_args: dict[str, Any] = None,
     ) -> None:
         """Creates a new instance of ``GeoJSONDataSet`` pointing to a concrete GeoJSON file
         on a specific filesystem fsspec.
@@ -116,7 +118,7 @@ def __init__(
         self._fs_open_args_load = _fs_open_args_load
         self._fs_open_args_save = _fs_open_args_save
 
-    def _load(self) -> Union[gpd.GeoDataFrame, Dict[str, gpd.GeoDataFrame]]:
+    def _load(self) -> Union[gpd.GeoDataFrame, dict[str, gpd.GeoDataFrame]]:
         load_path = get_filepath_str(self._get_load_path(), self._protocol)
         with self._fs.open(load_path, **self._fs_open_args_load) as fs_file:
             return gpd.read_file(fs_file, **self._load_args)
@@ -134,7 +136,7 @@ def _exists(self) -> bool:
             return False
         return self._fs.exists(load_path)
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         return {
             "filepath": self._filepath,
             "protocol": self._protocol,

kedro-datasets/kedro_datasets/holoviews/holoviews_writer.py

@@ -1,10 +1,11 @@
 """``HoloviewsWriter`` saves Holoviews objects as image file(s) to an underlying
 filesystem (e.g. local, S3, GCS)."""
+from __future__ import annotations
 
 import io
 from copy import deepcopy
 from pathlib import PurePosixPath
-from typing import Any, Dict, NoReturn, TypeVar
+from typing import Any, NoReturn, TypeVar
 
 import fsspec
 import holoviews as hv
@@ -37,15 +38,15 @@ class HoloviewsWriter(AbstractVersionedDataSet[HoloViews, NoReturn]):
 
     """
 
-    DEFAULT_SAVE_ARGS: Dict[str, Any] = {"fmt": "png"}
+    DEFAULT_SAVE_ARGS: dict[str, Any] = {"fmt": "png"}
 
     # pylint: disable=too-many-arguments
     def __init__(
         self,
         filepath: str,
-        fs_args: Dict[str, Any] = None,
-        credentials: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
+        fs_args: dict[str, Any] = None,
+        credentials: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
         version: Version = None,
     ) -> None:
         """Creates a new instance of ``HoloviewsWriter``.
@@ -97,7 +98,7 @@ def __init__(
         if save_args is not None:
             self._save_args.update(save_args)
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         return {
             "filepath": self._filepath,
             "protocol": self._protocol,

kedro-datasets/kedro_datasets/json/json_dataset.py

@@ -1,10 +1,12 @@
 """``JSONDataSet`` loads/saves data from/to a JSON file using an underlying
 filesystem (e.g.: local, S3, GCS). It uses native json to handle the JSON file.
 """
+from __future__ import annotations
+
 import json
 from copy import deepcopy
 from pathlib import PurePosixPath
-from typing import Any, Dict
+from typing import Any
 
 import fsspec
 from kedro.io.core import (
@@ -49,16 +51,16 @@ class JSONDataSet(AbstractVersionedDataSet[Any, Any]):
 
     """
 
-    DEFAULT_SAVE_ARGS: Dict[str, Any] = {"indent": 2}
+    DEFAULT_SAVE_ARGS: dict[str, Any] = {"indent": 2}
 
     # pylint: disable=too-many-arguments
     def __init__(
         self,
         filepath: str,
-        save_args: Dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
         version: Version = None,
-        credentials: Dict[str, Any] = None,
-        fs_args: Dict[str, Any] = None,
+        credentials: dict[str, Any] = None,
+        fs_args: dict[str, Any] = None,
     ) -> None:
         """Creates a new instance of ``JSONDataSet`` pointing to a concrete JSON file
         on a specific filesystem.
@@ -115,7 +117,7 @@ def __init__(
         self._fs_open_args_load = _fs_open_args_load
         self._fs_open_args_save = _fs_open_args_save
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         return {
             "filepath": self._filepath,
             "protocol": self._protocol,