Add metadata attribute to kedro.io datasets

kedro/io/cached_dataset.py

@@ -37,6 +37,7 @@ def __init__(
         dataset: Union[AbstractDataSet, Dict],
         version: Version = None,
         copy_mode: str = None,
+        metadata: Dict[str, Any] = None,
     ):
         """Creates a new instance of ``CachedDataSet`` pointing to the
         provided Python object.
@@ -50,6 +51,7 @@ def __init__(
             copy_mode: The copy mode used to copy the data. Possible
                 values are: "deepcopy", "copy" and "assign". If not
                 provided, it is inferred based on the data type.
+            metadata: Any arbitrary user metadata.
 
         Raises:
             ValueError: If the provided dataset is not a valid dict/YAML
@@ -65,6 +67,7 @@ def __init__(
                 "representation of the dataset, or the actual dataset object."
             )
         self._cache = MemoryDataSet(copy_mode=copy_mode)
+        self.metadata = metadata
 
     def _release(self) -> None:
         self._cache.release()

kedro/io/lambda_dataset.py

@@ -74,12 +74,14 @@ def _release(self) -> None:
         else:
             self.__release()
 
+    # pylint: disable=too-many-arguments
     def __init__(
         self,
         load: Optional[Callable[[], Any]],
         save: Optional[Callable[[Any], None]],
         exists: Callable[[], bool] = None,
         release: Callable[[], None] = None,
+        metadata: Dict[str, Any] = None,
     ):
         """Creates a new instance of ``LambdaDataSet`` with references to the
         required input/output data set methods.
@@ -89,6 +91,7 @@ def __init__(
             save: Method to save data to a data set.
             exists: Method to check whether output data already exists.
             release: Method to release any cached information.
+            metadata: Any arbitrary user metadata.
 
         Raises:
             DataSetError: If a method is specified, but is not a Callable.
@@ -111,3 +114,4 @@ def __init__(
         self.__save = save
         self.__exists = exists
         self.__release = release
+        self.metadata = metadata

kedro/io/memory_dataset.py

@@ -33,7 +33,9 @@ class MemoryDataSet(AbstractDataSet):
 
     """
 
-    def __init__(self, data: Any = _EMPTY, copy_mode: str = None):
+    def __init__(
+        self, data: Any = _EMPTY, copy_mode: str = None, metadata: Dict[str, Any] = None
+    ):
         """Creates a new instance of ``MemoryDataSet`` pointing to the
         provided Python object.
 
@@ -42,9 +44,11 @@ def __init__(self, data: Any = _EMPTY, copy_mode: str = None):
             copy_mode: The copy mode used to copy the data. Possible
                 values are: "deepcopy", "copy" and "assign". If not
                 provided, it is inferred based on the data type.
+            metadata: Any arbitrary user metadata.
         """
         self._data = _EMPTY
         self._copy_mode = copy_mode
+        self.metadata = metadata
         if data is not _EMPTY:
             self._save(data)
 

kedro/io/partitioned_dataset.py

@@ -139,6 +139,7 @@ def __init__(  # pylint: disable=too-many-arguments
         load_args: Dict[str, Any] = None,
         fs_args: Dict[str, Any] = None,
         overwrite: bool = False,
+        metadata: Dict[str, Any] = None,
     ):
         """Creates a new instance of ``PartitionedDataSet``.
 
@@ -177,6 +178,7 @@ def __init__(  # pylint: disable=too-many-arguments
             fs_args: Extra arguments to pass into underlying filesystem class constructor
                 (e.g. `{"project": "my-project"}` for ``GCSFileSystem``)
             overwrite: If True, any existing partitions will be removed.
+            metadata: Any arbitrary user metadata.
 
         Raises:
             DataSetError: If versioning is enabled for the underlying dataset.
@@ -191,6 +193,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._overwrite = overwrite
         self._protocol = infer_storage_options(self._path)["protocol"]
         self._partition_cache = Cache(maxsize=1)  # type: Cache
+        self.metadata = metadata
 
         dataset = dataset if isinstance(dataset, dict) else {"type": dataset}
         self._dataset_type, self._dataset_config = parse_dataset_definition(dataset)
@@ -381,6 +384,7 @@ def __init__(
         credentials: Dict[str, Any] = None,
         load_args: Dict[str, Any] = None,
         fs_args: Dict[str, Any] = None,
+        metadata: Dict[str, Any] = None,
     ):
 
         """Creates a new instance of ``IncrementalDataSet``.
@@ -427,6 +431,7 @@ def __init__(
                 the filesystem implementation.
             fs_args: Extra arguments to pass into underlying filesystem class constructor
                 (e.g. `{"project": "my-project"}` for ``GCSFileSystem``).
+            metadata: Any arbitrary user metadata.
 
         Raises:
             DataSetError: If versioning is enabled for the underlying dataset.
@@ -444,6 +449,7 @@ def __init__(
 
         self._checkpoint_config = self._parse_checkpoint_config(checkpoint)
         self._force_checkpoint = self._checkpoint_config.pop("force_checkpoint", None)
+        self.metadata = metadata
 
         comparison_func = self._checkpoint_config.pop("comparison_func", operator.gt)
         if isinstance(comparison_func, str):