🎨 🐛 Fix broken annotations in 3.9 due to #633 and upgrade all type annotations to python 3.9 standard

Author: Galileo-Galilei

CHANGELOG.md

@@ -7,6 +7,10 @@
 - :sparkles: Add the ``tracking.disable_tracking.disable_autologging`` configuration option in ``mlflow.yml `` to disable autologging by default. This simplify the workflow for Databricks users who have autologging activated by default, which conflicts with ``kedro-mlflow`` ([[#610](https://github.com/Galileo-Galilei/kedro-mlflow/issues/610)]).
 - :sparkles: Add ``tracking.experiment.create_experiment_kwargs.artifact_location`` and ``tracking.experiment.create_experiment_kwargs.tags`` configuration options in ``mlflow.yml `` to enable advanced configuration of mlflow experiment created at runtime by ``kedro-mlflow`` ([[#557](https://github.com/Galileo-Galilei/kedro-mlflow/issues/557)]).
 
+### Fixed
+
+- :bug: Fix type annotations introduced in [#633](https://github.com/Galileo-Galilei/kedro-mlflow/pull/633) which are not compatible with ``python==3.9``.
+
 ## [0.14.3] - 2025-02-17
 
 ### Added

docs/source/03_experiment_tracking/01_experiment_tracking/02_version_parameters.md

@@ -2,7 +2,7 @@
 
 ## Automatic parameters tracking
 
-Parameters tracking is automatic when the ``MlflowHook`` is added to [the hook list of the ``ProjectContext``](https://kedro-mlflow.readthedocs.io/en/latest/source/02_getting_started/01_installation/02_setup.html). The `mlflow.yml` configuration file has a parameter called ``flatten_dict_params`` which enables to [log as distinct parameters the (key, value) pairs of a ```Dict`` parameter](https://kedro-mlflow.readthedocs.io/en/latest/source/05_API/01_python_objects/02_Hooks.html).
+Parameters tracking is automatic when the ``MlflowHook`` is added to [the hook list of the ``ProjectContext``](https://kedro-mlflow.readthedocs.io/en/latest/source/02_getting_started/01_installation/02_setup.html). The `mlflow.yml` configuration file has a parameter called ``flatten_dict_params`` which enables to [log as distinct parameters the (key, value) pairs of a ```dict`` parameter](https://kedro-mlflow.readthedocs.io/en/latest/source/05_API/01_python_objects/02_Hooks.html).
 
 You **do not need any additional configuration** to benefit from parameters versioning.
 

docs/source/03_experiment_tracking/01_experiment_tracking/05_version_metrics.md

@@ -172,7 +172,7 @@ my_model_metrics:
 Let assume that you have node which doesn't have any inputs and returns dictionary with metrics to log:
 
 ```python
-def metrics_node() -> Dict[str, Union[float, List[float]]]:
+def metrics_node() -> dict[str, Union[float, list[float]]]:
     return {
         "metric1": {"value": 1.1, "step": 1},
         "metric2": [{"value": 1.1, "step": 1}, {"value": 1.2, "step": 2}],
@@ -181,8 +181,8 @@ def metrics_node() -> Dict[str, Union[float, List[float]]]:
 
 As you can see above, ``kedro_mlflow.io.metrics.MlflowMetricsHistoryDataset`` can take metrics as:
 
-- ``Dict[str, key]``
-- ``List[Dict[str, key]]``
+- ``[str, key]``
+- ``list[[str, key]]``
 
 To store metrics we need to define metrics dataset in Kedro Catalog:
 

docs/source/04_pipeline_as_model/01_pipeline_as_custom_model/02_scikit_learn_like_pipeline.md

@@ -23,7 +23,7 @@ You can configure your project as follows:
     from kedro_mlflow_tutorial.pipelines.ml_app.pipeline import create_ml_pipeline
 
 
-    def register_pipelines(self) -> Dict[str, Pipeline]:
+    def register_pipelines(self) -> [str, Pipeline]:
         ml_pipeline = create_ml_pipeline()
         training_pipeline_ml = pipeline_ml_factory(
             training=ml_pipeline.only_nodes_with_tags(

docs/source/04_pipeline_as_model/02_framework_ml/03_framework_solutions.md

@@ -24,7 +24,7 @@ from kedro_mlflow_tutorial.pipelines.ml_app.pipeline import create_ml_pipeline
 
 class ProjectHooks:
     @hook_impl
-    def register_pipelines(self) -> Dict[str, Pipeline]:
+    def register_pipelines(self) -> [str, Pipeline]:
         ml_pipeline = create_ml_pipeline()
 
         # convert your two pipelines to a PipelinML object

docs/source/05_API/01_python_objects/01_Datasets.md

@@ -62,8 +62,8 @@ The ``MlflowModelTrackingDataset`` accepts the following arguments:
 - run_id (Optional[str], optional): MLflow run ID to use to load the model from or save the model to. It plays the same role as "filepath" for standard mlflow datasets. Defaults to None.
 - artifact_path (str, optional): the run relative path to the model.
 - pyfunc_workflow (str, optional): Either `python_model` or `loader_module`.See [mlflow workflows](https://www.mlflow.org/docs/latest/python_api/mlflow.pyfunc.html#workflows).
-- load_args (Dict[str, Any], optional): Arguments to `load_model` function from specified `flavor`. Defaults to None.
-- save_args (Dict[str, Any], optional): Arguments to `log_model` function from specified `flavor`. Defaults to None.
+- load_args (dict[str, Any], optional): Arguments to `load_model` function from specified `flavor`. Defaults to None.
+- save_args (dict[str, Any], optional): Arguments to `log_model` function from specified `flavor`. Defaults to None.
 
 You can either only specify the flavor:
 
@@ -122,8 +122,8 @@ The ``MlflowModelLocalFileSystemDataset`` accepts the following arguments:
 - flavor (str): Built-in or custom MLflow model flavor module. Must be Python-importable.
 - filepath (str): Path to store the dataset locally.
 - pyfunc_workflow (str, optional): Either `python_model` or `loader_module`. See [mlflow workflows](https://www.mlflow.org/docs/latest/python_api/mlflow.pyfunc.html#workflows).
-- load_args (Dict[str, Any], optional): Arguments to `load_model` function from specified `flavor`. Defaults to None.
-- save_args (Dict[str, Any], optional): Arguments to `save_model` function from specified `flavor`. Defaults to None.
+- load_args (dict[str, Any], optional): Arguments to `load_model` function from specified `flavor`. Defaults to None.
+- save_args (dict[str, Any], optional): Arguments to `save_model` function from specified `flavor`. Defaults to None.
 - version (Version, optional): Kedro version to use. Defaults to None.
 
 The use is very similar to ``MlflowModelTrackingDataset``, but you have to specify a local ``filepath`` instead of a `run_id`:
@@ -168,7 +168,7 @@ The ``MlflowModelRegistryDataset`` accepts the following arguments:
 - ``alias`` (str): A valid alias, which is used instead of stage to filter model since mlflow 2.9.0. Will raise an error if both ``stage_or_version`` and ``alias`` are provided.
 - ``flavor`` (str): Built-in or custom MLflow model flavor module. Must be Python-importable.
 - ``pyfunc_workflow`` (str, optional): Either `python_model` or `loader_module`. See [mlflow workflows](https://www.mlflow.org/docs/latest/python_api/mlflow.pyfunc.html#workflows).
-- ``load_args`` (Dict[str, Any], optional): Arguments to `load_model` function from specified `flavor`. Defaults to None.
+- ``load_args`` (dict[str, Any], optional): Arguments to `load_model` function from specified `flavor`. Defaults to None.
 
 We assume you have registered a mlflow model first, either [with the ``MlflowClient``](https://mlflow.org/docs/latest/model-registry.html#adding-an-mlflow-model-to-the-model-registry) or [within the mlflow ui](https://mlflow.org/docs/latest/model-registry.html#ui-workflow), e.g. :
 

docs/source/05_API/01_python_objects/03_Pipelines.md

@@ -14,7 +14,7 @@ Example within kedro template:
 from PYTHON_PACKAGE.pipelines import data_science as ds
 
 
-def create_pipelines(**kwargs) -> Dict[str, Pipeline]:
+def create_pipelines(**kwargs) -> dict[str, Pipeline]:
     data_science_pipeline = ds.create_pipeline()
     training_pipeline = pipeline_ml_factory(
         training=data_science_pipeline.only_nodes_with_tags(

kedro_mlflow/framework/cli/cli.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from platform import python_version
 from tempfile import TemporaryDirectory
-from typing import Dict, Optional, Union
+from typing import Optional, Union
 
 import click
 import mlflow
@@ -295,7 +295,7 @@ def modelify(
     flag_infer_input_example: Optional[bool],
     run_id: Optional[str],
     run_name: Optional[str],
-    copy_mode: Optional[Union[str, Dict[str, str]]],
+    copy_mode: Optional[Union[str, dict[str, str]]],
     artifact_path: str,
     code_path: str,
     conda_env: Optional[str],

kedro_mlflow/framework/hooks/mlflow_hook.py

@@ -3,7 +3,7 @@
 from logging import Logger, getLogger
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Any, Dict, Union
+from typing import Any, Union
 
 import mlflow
 from kedro.config import MissingConfigException
@@ -145,9 +145,9 @@ def after_context_created(
     def after_catalog_created(
         self,
         catalog: DataCatalog,
-        conf_catalog: Dict[str, Any],
-        conf_creds: Dict[str, Any],
-        feed_dict: Dict[str, Any],
+        conf_catalog: dict[str, Any],
+        conf_creds: dict[str, Any],
+        feed_dict: dict[str, Any],
         save_version: str,
         load_versions: str,
     ):
@@ -197,7 +197,7 @@ def after_catalog_created(
 
     @hook_impl
     def before_pipeline_run(
-        self, run_params: Dict[str, Any], pipeline: Pipeline, catalog: DataCatalog
+        self, run_params: dict[str, Any], pipeline: Pipeline, catalog: DataCatalog
     ) -> None:
         """Hook to be invoked before a pipeline runs.
         Args:
@@ -208,14 +208,14 @@ def before_pipeline_run(
                     "project_path": str,
                     "env": str,
                     "kedro_version": str,
-                    "tags": Optional[List[str]],
-                    "from_nodes": Optional[List[str]],
-                    "to_nodes": Optional[List[str]],
-                    "node_names": Optional[List[str]],
-                    "from_inputs": Optional[List[str]],
-                    "load_versions": Optional[List[str]],
+                    "tags": Optional[list[str]],
+                    "from_nodes": Optional[list[str]],
+                    "to_nodes": Optional[list[str]],
+                    "node_names": Optional[list[str]],
+                    "from_inputs": Optional[list[str]],
+                    "load_versions": Optional[list[str]],
                     "pipeline_name": str,
-                    "extra_params": Optional[Dict[str, Any]],
+                    "extra_params": Optional[dict[str, Any]],
                 }
             pipeline: The ``Pipeline`` that will be run.
             catalog: The ``DataCatalog`` to be used during the run.
@@ -282,7 +282,7 @@ def before_pipeline_run(
 
     @hook_impl
     def before_node_run(
-        self, node: Node, catalog: DataCatalog, inputs: Dict[str, Any], is_async: bool
+        self, node: Node, catalog: DataCatalog, inputs: dict[str, Any], is_async: bool
     ) -> None:
         """Hook to be invoked before a node runs.
         This hook logs all the parameters of the nodes in mlflow.
@@ -342,7 +342,7 @@ def before_node_run(
             for k, v in params_inputs.items():
                 self._log_param(k, v)
 
-    def _log_param(self, name: str, value: Union[Dict, int, bool, str]) -> None:
+    def _log_param(self, name: str, value: Union[dict, int, bool, str]) -> None:
         str_value = str(value)
         str_value_length = len(str_value)
         if str_value_length <= MAX_PARAM_VAL_LENGTH:
@@ -368,7 +368,7 @@ def _log_param(self, name: str, value: Union[Dict, int, bool, str]) -> None:
     @hook_impl
     def after_pipeline_run(
         self,
-        run_params: Dict[str, Any],
+        run_params: dict[str, Any],
         pipeline: Pipeline,
         catalog: DataCatalog,
     ) -> None:
@@ -381,14 +381,14 @@ def after_pipeline_run(
                     "project_path": str,
                     "env": str,
                     "kedro_version": str,
-                    "tags": Optional[List[str]],
-                    "from_nodes": Optional[List[str]],
-                    "to_nodes": Optional[List[str]],
-                    "node_names": Optional[List[str]],
-                    "from_inputs": Optional[List[str]],
-                    "load_versions": Optional[List[str]],
+                    "tags": Optional[list[str]],
+                    "from_nodes": Optional[list[str]],
+                    "to_nodes": Optional[list[str]],
+                    "node_names": Optional[list[str]],
+                    "from_inputs": Optional[list[str]],
+                    "load_versions": Optional[list[str]],
                     "pipeline_name": str,
-                    "extra_params": Optional[Dict[str, Any]],
+                    "extra_params": Optional[dict[str, Any]],
                 }
             pipeline: The ``Pipeline`` that was run.
             catalog: The ``DataCatalog`` used during the run.
@@ -451,7 +451,7 @@ def after_pipeline_run(
     def on_pipeline_error(
         self,
         error: Exception,
-        run_params: Dict[str, Any],
+        run_params: dict[str, Any],
         pipeline: Pipeline,
         catalog: DataCatalog,
     ):
@@ -467,14 +467,14 @@ def on_pipeline_error(
                      "project_path": str,
                      "env": str,
                      "kedro_version": str,
-                     "tags": Optional[List[str]],
-                     "from_nodes": Optional[List[str]],
-                     "to_nodes": Optional[List[str]],
-                     "node_names": Optional[List[str]],
-                     "from_inputs": Optional[List[str]],
-                     "load_versions": Optional[List[str]],
+                     "tags": Optional[list[str]],
+                     "from_nodes": Optional[list[str]],
+                     "to_nodes": Optional[list[str]],
+                     "node_names": Optional[list[str]],
+                     "from_inputs": Optional[list[str]],
+                     "load_versions": Optional[list[str]],
                      "pipeline_name": str,
-                     "extra_params": Optional[Dict[str, Any]]
+                     "extra_params": Optional[dict[str, Any]]
                    }
             pipeline: (Not used) The ``Pipeline`` that will was run.
             catalog: (Not used) The ``DataCatalog`` used during the run.

kedro_mlflow/framework/hooks/utils.py

@@ -1,5 +1,3 @@
-from typing import Dict
-
 from kedro_mlflow.config.kedro_mlflow_config import KedroMlflowConfig
 
 
@@ -43,7 +41,7 @@ def _generate_kedro_command(
     return kedro_cmd
 
 
-def _flatten_dict(d: Dict, recursive: bool = True, sep: str = ".") -> Dict:
+def _flatten_dict(d: dict, recursive: bool = True, sep: str = ".") -> dict:
     def expand(key, value):
         if isinstance(value, dict):
             new_value = (

kedro_mlflow/io/artifacts/mlflow_artifact_dataset.py

@@ -1,6 +1,6 @@
 import shutil
 from pathlib import Path
-from typing import Any, Dict, Union
+from typing import Any, Optional, Union
 
 import mlflow
 from kedro.io import AbstractVersionedDataset
@@ -15,11 +15,11 @@ class MlflowArtifactDataset(AbstractVersionedDataset):
 
     def __new__(
         cls,
-        dataset: Union[str, Dict],
+        dataset: Union[str, dict],
         run_id: str = None,
         artifact_path: str = None,
-        credentials: Dict[str, Any] = None,
-        metadata: Dict[str, Any] | None = None,
+        credentials: dict[str, Any] = None,
+        metadata: Optional[dict[str, Any]] = None,
     ):
         dataset_obj, dataset_args = parse_dataset_definition(config=dataset)
 
@@ -169,7 +169,7 @@ def _save(self, data: Any) -> None:  # pragma: no cover
         """
         pass
 
-    def _describe(self) -> Dict[str, Any]:  # pragma: no cover
+    def _describe(self) -> dict[str, Any]:  # pragma: no cover
         """
         MlflowArtifactDataset is a factory for DataSet
         and consequently does not implements abtracts methods

kedro_mlflow/io/metrics/mlflow_abstract_metric_dataset.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Union
+from typing import Any, Optional, Union
 
 import mlflow
 from kedro.io import AbstractDataset
@@ -10,9 +10,9 @@ def __init__(
         self,
         key: str = None,
         run_id: str = None,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
-        metadata: Dict[str, Any] | None = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
+        metadata: Optional[dict[str, Any]] = None,
     ):
         """Initialise MlflowMetricsHistoryDataset.
 
@@ -81,11 +81,11 @@ def _exists(self) -> bool:
         flag_exist = self.key in run.data.metrics.keys() if run else False
         return flag_exist
 
-    def _describe(self) -> Dict[str, Any]:
+    def _describe(self) -> dict[str, Any]:
         """Describe MLflow metrics dataset.
 
         Returns:
-            Dict[str, Any]: Dictionary with MLflow metrics dataset description.
+            dict[str, Any]: dictionary with MLflow metrics dataset description.
         """
         return {
             "key": self.key,

kedro_mlflow/io/metrics/mlflow_metric_dataset.py

@@ -1,5 +1,5 @@
 from copy import deepcopy
-from typing import Any, Dict
+from typing import Any, Optional
 
 from mlflow.tracking import MlflowClient
 
@@ -16,9 +16,9 @@ def __init__(
         self,
         key: str = None,
         run_id: str = None,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
-        metadata: Dict[str, Any] | None = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
+        metadata: Optional[dict[str, Any]] = None,
     ):
         """Initialise MlflowMetricDataset.
         Args:

kedro_mlflow/io/metrics/mlflow_metric_history_dataset.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List, Union
+from typing import Any, Optional, Union
 
 from mlflow.tracking import MlflowClient
 
@@ -12,9 +12,9 @@ def __init__(
         self,
         key: str = None,
         run_id: str = None,
-        load_args: Dict[str, Any] = None,
-        save_args: Dict[str, Any] = None,
-        metadata: Dict[str, Any] | None = None,
+        load_args: dict[str, Any] = None,
+        save_args: dict[str, Any] = None,
+        metadata: Optional[dict[str, Any]] = None,
     ):
         """Initialise MlflowMetricDataset.
         Args:
@@ -51,7 +51,7 @@ def _load(self):
 
     def _save(
         self,
-        data: Union[List[int], Dict[int, float], List[Dict[str, Union[float, str]]]],
+        data: Union[list[int], dict[int, float], list[dict[str, Union[float, str]]]],
     ):
         if self._logging_activated:
             self._validate_run_id()