Repository: Avaiga/taipy
Pull request: Feature/#1196 - Add download and upload api for file-based datanodes
Resolves #1196

In this PR, 2 new APIs are open for file-based DataNode:
- `_get_downloadable_path()` that returns the path to the downloadable data file
- `_upload()` that check the uploaded file and replace the current data file

Problem statement:
Data node Viewer - Add file download/upload capability for file based data nodes
**Description**
When a data node is stored as a file, it would be handy for the end user to add a way to download or upload the file from the data node viewer easily.

This should help the end user understand its data, particularly when data is corrupted or an unexpected behavior happens.

This capability could be enabled/disabled by the developer through a property. By default, it should be enabled.

**Proposal**
1. The download could be accessible through a "Download" button (eventually, an icon when the element width is too small).
2. The upload could be accessible through an "Upload" file selector (eventually, an icon when the element width is too small). When a file is passed to the Upload mechanism, Taipy should:
    - Check the file (a check property should let the developer provide a custom function).
    - Write the file at the data node path location. Eventually, overwrite it if a file already exists.
    - track the data node edition

**Usage**
```python
def custom_check(state, file_path) -> bool:
    # Apply the check logic
    return True


<| {selected_datanode} | datanode | file_download=false | file_upload=true | upload_check=custom_check |>
```


- [ ] Create a ticket for the Gui part (buttons into the datanode viewer)

Hints from issue discussion:
Seems related to this https://github.com/Avaiga/taipy/issues/427
There are a few use-cases I think we need to clarify before actually implement the feature. I will only mainly on the Core side of the problem here, and error handling on GUI as well.

1. If the datanode is not file-based, what happens?
   - Raise an error on the GUI component directly? (not sure if it's possible), or
   - Notify an error from Core to GUI

2. For most file-based datanodes, we can return the file or folder directly when downloading. But on ExcelDataNode, what if the DataNode can only access a few sheets of the file, we return the original file or a new file with only the defined sheets?

3. For uploading data, how do we handle invalid data?
   - We can check for the data before writing,
   - Or better yet, we can try to write the data and catch any exception to raise an error. 
   - However, most issues related to wrong data will only be raised when executing jobs, so it's hard to detect the problem at uploading step.

1. Make the button inactive.
2. I would guess the new file (not sure)
3. I would say it is on the developer side, not ours. Make sure their tasks are compliant with invalid data or do the right transformation.
These are good questions.
1. I agree with @FlorianJacta. We simply deactivate the button.
2. Good question. What I had in mind originally was just asking the data node path to Core, and then simply download the file, without any preprocessing. 
If we need to go through the read method to understand what should be exported, the functionality is not the same. In this case, I believe a table export feature to export the content of a table is more suitable and could apply to SQL data nodes as well.
3. Here again I was originally thinking of a simple file replacement. That means the responsibility is on our users.
_The end-user_: The end user should take the responsibility to ensure the validity of the data. We can imagine adding an info message (like a mouseover text or something else) to let him know that he is responsible for checking the file format (pickle, excel, csv) and eventually the data format (column names). This message could be specific to each data node type.
_The developer_: We should expose a way for the developer to pass a checker method to the control. THe checker can validate the file content once uploaded. If the checker passes the file can be replaced.

@MariahLeone Any thoughts about that.

Commit message: feat: add download and upload api for file-based datanodes

Patch to review:
diff --git a/taipy/core/data/_file_datanode_mixin.py b/taipy/core/data/_file_datanode_mixin.py
index 4521922d25..cbab6614cf 100644
--- a/taipy/core/data/_file_datanode_mixin.py
+++ b/taipy/core/data/_file_datanode_mixin.py
@@ -14,7 +14,7 @@
 import shutil
 from datetime import datetime
 from os.path import isfile
-from typing import Any, Dict, Optional
+from typing import Any, Callable, Dict, Optional
 
 from taipy.config.config import Config
 
@@ -92,3 +92,50 @@ def _migrate_path(self, storage_type, old_path) -> str:
         if os.path.exists(old_path):
             shutil.move(old_path, new_path)
         return new_path
+
+    def _get_downloadable_path(self) -> str:
+        """Get the downloadable path of the file data of the data node.
+
+        Returns:
+            The downloadable path of the file data of the data node if it exists, otherwise an empty string.
+        """
+        if os.path.exists(self.path) and isfile(self._path):
+            return self.path
+
+        return ""
+
+    def _upload(self, path: str, upload_checker: Optional[Callable[[str, Any], bool]] = None) -> bool:
+        """Upload a csv file data to the data node.
+
+        Parameters:
+            path (str): The path of the file to upload to the data node.
+            upload_checker (Optional[Callable[[str, Any], bool]]): A function to check if the upload is allowed.
+                The function takes the title of the upload data and the data itself as arguments and returns
+                True if the upload is allowed, otherwise False.
+
+        Returns:
+            True if the upload was successful, otherwise False.
+        """
+        from ._data_manager_factory import _DataManagerFactory
+
+        upload_path = pathlib.Path(path)
+
+        if upload_checker is not None:
+            try:
+                upload_data = self._read_from_path(upload_path.name)
+            except Exception:
+                return False
+
+            if not upload_checker(upload_path.name, upload_data):
+                return False
+
+        shutil.copy(upload_path, self.path)
+
+        self.track_edit(timestamp=datetime.now())  # type: ignore[attr-defined]
+        self.unlock_edit()  # type: ignore[attr-defined]
+        _DataManagerFactory._build_manager()._set(self)  # type: ignore[arg-type]
+
+        return True
+
+    def _read_from_path(self, path: Optional[str] = None, **read_kwargs) -> Any:
+        raise NotImplementedError
diff --git a/taipy/core/data/csv.py b/taipy/core/data/csv.py
index 472f77b4dc..26410f45a1 100644
--- a/taipy/core/data/csv.py
+++ b/taipy/core/data/csv.py
@@ -27,7 +27,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class CSVDataNode(DataNode, _FileDataNodeMixin, _TabularDataNodeMixin):
+class CSVDataNode(_FileDataNodeMixin, DataNode, _TabularDataNodeMixin):
     """Data Node stored as a CSV file.
 
     Attributes:
@@ -137,14 +137,20 @@ def storage_type(cls) -> str:
         return cls.__STORAGE_TYPE
 
     def _read(self):
+        return self._read_from_path()
+
+    def _read_from_path(self, path: Optional[str] = None, **read_kwargs) -> Any:
+        if not path:
+            path = self._path
+
         if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_PANDAS:
-            return self._read_as_pandas_dataframe()
+            return self._read_as_pandas_dataframe(path=path)
         if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_NUMPY:
-            return self._read_as_numpy()
-        return self._read_as()
+            return self._read_as_numpy(path=path)
+        return self._read_as(path=path)
 
-    def _read_as(self):
-        with open(self._path, encoding=self.properties[self.__ENCODING_KEY]) as csvFile:
+    def _read_as(self, path: str):
+        with open(path, encoding=self.properties[self.__ENCODING_KEY]) as csvFile:
             if self.properties[self._HAS_HEADER_PROPERTY]:
                 reader = csv.DictReader(csvFile)
             else:
@@ -152,23 +158,26 @@ def _read_as(self):
 
             return [self._decoder(line) for line in reader]
 
-    def _read_as_numpy(self) -> np.ndarray:
-        return self._read_as_pandas_dataframe().to_numpy()
+    def _read_as_numpy(self, path: str) -> np.ndarray:
+        return self._read_as_pandas_dataframe(path=path).to_numpy()
 
     def _read_as_pandas_dataframe(
-        self, usecols: Optional[List[int]] = None, column_names: Optional[List[str]] = None
+        self,
+        path: str,
+        usecols: Optional[List[int]] = None,
+        column_names: Optional[List[str]] = None,
     ) -> pd.DataFrame:
         try:
             if self.properties[self._HAS_HEADER_PROPERTY]:
                 if column_names:
-                    return pd.read_csv(self._path, encoding=self.properties[self.__ENCODING_KEY])[column_names]
-                return pd.read_csv(self._path, encoding=self.properties[self.__ENCODING_KEY])
+                    return pd.read_csv(path, encoding=self.properties[self.__ENCODING_KEY])[column_names]
+                return pd.read_csv(path, encoding=self.properties[self.__ENCODING_KEY])
             else:
                 if usecols:
                     return pd.read_csv(
-                        self._path, encoding=self.properties[self.__ENCODING_KEY], header=None, usecols=usecols
+                        path, encoding=self.properties[self.__ENCODING_KEY], header=None, usecols=usecols
                     )
-                return pd.read_csv(self._path, encoding=self.properties[self.__ENCODING_KEY], header=None)
+                return pd.read_csv(path, encoding=self.properties[self.__ENCODING_KEY], header=None)
         except pd.errors.EmptyDataError:
             return pd.DataFrame()
 
diff --git a/taipy/core/data/data_node.py b/taipy/core/data/data_node.py
index 32f7067a7d..ef91035cd7 100644
--- a/taipy/core/data/data_node.py
+++ b/taipy/core/data/data_node.py
@@ -14,7 +14,7 @@
 import uuid
 from abc import abstractmethod
 from datetime import datetime, timedelta
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
 
 import networkx as nx
 
@@ -373,6 +373,28 @@ def _get_last_modified_datetime(cls, path: Optional[str] = None) -> Optional[dat
     def storage_type(cls) -> str:
         raise NotImplementedError
 
+    def _get_downloadable_path(self) -> str:
+        """Get the downloadable path of the file data of the data node.
+
+        Returns:
+            The downloadable path of the file data of the data node.
+        """
+        raise NotImplementedError
+
+    def _upload(self, path: str, upload_checker: Optional[Callable[[str, Any], bool]] = None) -> bool:
+        """Upload the file data of the data node.
+
+        Parameters:
+            path (str): The path of the file to upload to the data node.
+            upload_checker (Optional[Callable[[str, Any], bool]]): A function to check if the upload is allowed.
+                The function takes the title of the upload data and the data itself as arguments and returns
+                True if the upload is allowed, otherwise False.
+
+        Returns:
+            True if the upload was successful, otherwise False.
+        """
+        raise NotImplementedError
+
     def read_or_raise(self) -> Any:
         """Read the data referenced by this data node.
 
diff --git a/taipy/core/data/excel.py b/taipy/core/data/excel.py
index d9ee54fa31..61777eaed4 100644
--- a/taipy/core/data/excel.py
+++ b/taipy/core/data/excel.py
@@ -28,7 +28,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class ExcelDataNode(DataNode, _FileDataNodeMixin, _TabularDataNodeMixin):
+class ExcelDataNode(_FileDataNodeMixin, DataNode, _TabularDataNodeMixin):
     """Data Node stored as an Excel file.
 
     The Excel file format is _xlsx_.
@@ -150,15 +150,21 @@ def _check_exposed_type(exposed_type):
                 _TabularDataNodeMixin._check_exposed_type(t)
 
     def _read(self):
+        return self._read_from_path()
+
+    def _read_from_path(self, path: Optional[str] = None, **read_kwargs) -> Any:
+        if not path:
+            path = self._path
+
         if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_PANDAS:
-            return self._read_as_pandas_dataframe()
+            return self._read_as_pandas_dataframe(path=path)
         if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_NUMPY:
-            return self._read_as_numpy()
-        return self._read_as()
+            return self._read_as_numpy(path=path)
+        return self._read_as(path=path)
 
-    def _read_as(self):
+    def _read_as(self, path: str):
         try:
-            excel_file = load_workbook(self._path)
+            excel_file = load_workbook(path)
             exposed_type = self.properties[self._EXPOSED_TYPE_PROPERTY]
             work_books = {}
             sheet_names = excel_file.sheetnames
@@ -171,7 +177,7 @@ def _read_as(self):
 
             for sheet_name in provided_sheet_names:
                 if sheet_name not in sheet_names:
-                    raise NonExistingExcelSheet(sheet_name, self._path)
+                    raise NonExistingExcelSheet(sheet_name, path)
 
             if isinstance(exposed_type, List):
                 if len(provided_sheet_names) != len(self.properties[self._EXPOSED_TYPE_PROPERTY]):
@@ -192,9 +198,9 @@ def _read_as(self):
 
                     if isinstance(sheet_exposed_type, str):
                         if sheet_exposed_type == self._EXPOSED_TYPE_NUMPY:
-                            work_books[sheet_name] = self._read_as_pandas_dataframe(sheet_name).to_numpy()
+                            work_books[sheet_name] = self._read_as_pandas_dataframe(path, sheet_name).to_numpy()
                         elif sheet_exposed_type == self._EXPOSED_TYPE_PANDAS:
-                            work_books[sheet_name] = self._read_as_pandas_dataframe(sheet_name)
+                            work_books[sheet_name] = self._read_as_pandas_dataframe(path, sheet_name)
                         continue
 
                 res = [[col.value for col in row] for row in work_sheet.rows]
@@ -214,14 +220,16 @@ def _read_as(self):
 
         return work_books
 
-    def _read_as_numpy(self):
-        sheets = self._read_as_pandas_dataframe()
+    def _read_as_numpy(self, path: str):
+        sheets = self._read_as_pandas_dataframe(path=path)
         if isinstance(sheets, dict):
             return {sheet_name: df.to_numpy() for sheet_name, df in sheets.items()}
         return sheets.to_numpy()
 
-    def _do_read_excel(self, sheet_names, kwargs) -> Union[Dict[Union[int, str], pd.DataFrame], pd.DataFrame]:
-        return pd.read_excel(self._path, sheet_name=sheet_names, **kwargs)
+    def _do_read_excel(
+        self, path: str, sheet_names, kwargs
+    ) -> Union[Dict[Union[int, str], pd.DataFrame], pd.DataFrame]:
+        return pd.read_excel(path, sheet_name=sheet_names, **kwargs)
 
     def __get_sheet_names_and_header(self, sheet_names):
         kwargs = {}
@@ -231,10 +239,12 @@ def __get_sheet_names_and_header(self, sheet_names):
             kwargs["header"] = None
         return sheet_names, kwargs
 
-    def _read_as_pandas_dataframe(self, sheet_names=None) -> Union[Dict[Union[int, str], pd.DataFrame], pd.DataFrame]:
+    def _read_as_pandas_dataframe(
+        self, path: str, sheet_names=None
+    ) -> Union[Dict[Union[int, str], pd.DataFrame], pd.DataFrame]:
         sheet_names, kwargs = self.__get_sheet_names_and_header(sheet_names)
         try:
-            return self._do_read_excel(sheet_names, kwargs)
+            return self._do_read_excel(path, sheet_names, kwargs)
         except pd.errors.EmptyDataError:
             return pd.DataFrame()
 
diff --git a/taipy/core/data/json.py b/taipy/core/data/json.py
index f361bc9e60..e64f8b62ad 100644
--- a/taipy/core/data/json.py
+++ b/taipy/core/data/json.py
@@ -25,7 +25,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class JSONDataNode(DataNode, _FileDataNodeMixin):
+class JSONDataNode(_FileDataNodeMixin, DataNode):
     """Data Node stored as a JSON file.
 
     Attributes:
@@ -150,7 +150,13 @@ def decoder(self, decoder: json.JSONDecoder):
         self.properties[self._DECODER_KEY] = decoder
 
     def _read(self):
-        with open(self._path, "r", encoding=self.properties[self.__ENCODING_KEY]) as f:
+        return self._read_from_path()
+
+    def _read_from_path(self, path: Optional[str] = None, **read_kwargs) -> Any:
+        if not path:
+            path = self._path
+
+        with open(path, "r", encoding=self.properties[self.__ENCODING_KEY]) as f:
             return json.load(f, cls=self._decoder)
 
     def _append(self, data: Any):
diff --git a/taipy/core/data/parquet.py b/taipy/core/data/parquet.py
index 9c80856cbd..9f3672b695 100644
--- a/taipy/core/data/parquet.py
+++ b/taipy/core/data/parquet.py
@@ -28,7 +28,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class ParquetDataNode(DataNode, _FileDataNodeMixin, _TabularDataNodeMixin):
+class ParquetDataNode(_FileDataNodeMixin, DataNode, _TabularDataNodeMixin):
     """Data Node stored as a Parquet file.
 
     Attributes:
@@ -178,18 +178,43 @@ def storage_type(cls) -> str:
         return cls.__STORAGE_TYPE
 
     def _read(self):
-        return self.read_with_kwargs()
+        return self._read_from_path()
 
-    def _read_as(self, read_kwargs: Dict):
+    def _read_from_path(self, path: Optional[str] = None, **read_kwargs) -> Any:
+        if not path:
+            path = self._path
+
+        # return None if data was never written
+        if not self.last_edit_date:
+            self._DataNode__logger.warning(
+                f"Data node {self.id} from config {self.config_id} is being read but has never been written."
+            )
+            return None
+
+        kwargs = self.properties[self.__READ_KWARGS_PROPERTY]
+        kwargs.update(
+            {
+                self.__ENGINE_PROPERTY: self.properties[self.__ENGINE_PROPERTY],
+            }
+        )
+        kwargs.update(read_kwargs)
+
+        if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_PANDAS:
+            return self._read_as_pandas_dataframe(path, kwargs)
+        if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_NUMPY:
+            return self._read_as_numpy(path, kwargs)
+        return self._read_as(path, kwargs)
+
+    def _read_as(self, path: str, read_kwargs: Dict):
         custom_class = self.properties[self._EXPOSED_TYPE_PROPERTY]
-        list_of_dicts = self._read_as_pandas_dataframe(read_kwargs).to_dict(orient="records")
+        list_of_dicts = self._read_as_pandas_dataframe(path, read_kwargs).to_dict(orient="records")
         return [custom_class(**dct) for dct in list_of_dicts]
 
-    def _read_as_numpy(self, read_kwargs: Dict) -> np.ndarray:
-        return self._read_as_pandas_dataframe(read_kwargs).to_numpy()
+    def _read_as_numpy(self, path: str, read_kwargs: Dict) -> np.ndarray:
+        return self._read_as_pandas_dataframe(path, read_kwargs).to_numpy()
 
-    def _read_as_pandas_dataframe(self, read_kwargs: Dict) -> pd.DataFrame:
-        return pd.read_parquet(self._path, **read_kwargs)
+    def _read_as_pandas_dataframe(self, path: str, read_kwargs: Dict) -> pd.DataFrame:
+        return pd.read_parquet(path, **read_kwargs)
 
     def _append(self, data: Any):
         self.write_with_kwargs(data, engine="fastparquet", append=True)
@@ -233,23 +258,4 @@ def read_with_kwargs(self, **read_kwargs):
             **read_kwargs (dict[str, any]): The keyword arguments passed to the function
                 `pandas.read_parquet()`.
         """
-        # return None if data was never written
-        if not self.last_edit_date:
-            self._DataNode__logger.warning(
-                f"Data node {self.id} from config {self.config_id} is being read but has never been written."
-            )
-            return None
-
-        kwargs = self.properties[self.__READ_KWARGS_PROPERTY]
-        kwargs.update(
-            {
-                self.__ENGINE_PROPERTY: self.properties[self.__ENGINE_PROPERTY],
-            }
-        )
-        kwargs.update(read_kwargs)
-
-        if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_PANDAS:
-            return self._read_as_pandas_dataframe(kwargs)
-        if self.properties[self._EXPOSED_TYPE_PROPERTY] == self._EXPOSED_TYPE_NUMPY:
-            return self._read_as_numpy(kwargs)
-        return self._read_as(kwargs)
+        return self._read_from_path(**read_kwargs)
diff --git a/taipy/core/data/pickle.py b/taipy/core/data/pickle.py
index 7c7693ff68..9f363de00f 100644
--- a/taipy/core/data/pickle.py
+++ b/taipy/core/data/pickle.py
@@ -11,7 +11,7 @@
 
 import pickle
 from datetime import datetime, timedelta
-from typing import List, Optional, Set
+from typing import Any, List, Optional, Set
 
 from taipy.config.common.scope import Scope
 
@@ -22,7 +22,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class PickleDataNode(DataNode, _FileDataNodeMixin):
+class PickleDataNode(_FileDataNodeMixin, DataNode):
     """Data Node stored as a pickle file.
 
     Attributes:
@@ -116,7 +116,13 @@ def storage_type(cls) -> str:
         return cls.__STORAGE_TYPE
 
     def _read(self):
-        with open(self._path, "rb") as pf:
+        return self._read_from_path()
+
+    def _read_from_path(self, path: Optional[str] = None, **read_kwargs) -> Any:
+        if not path:
+            path = self._path
+
+        with open(path, "rb") as pf:
             return pickle.load(pf)
 
     def _write(self, data):
diff --git a/tests/core/data/test_csv_data_node.py b/tests/core/data/test_csv_data_node.py
index 0ed8ede596..069cc408a0 100644
--- a/tests/core/data/test_csv_data_node.py
+++ b/tests/core/data/test_csv_data_node.py
@@ -15,8 +15,10 @@
 from datetime import datetime
 from time import sleep
 
+import numpy as np
 import pandas as pd
 import pytest
+from pandas.testing import assert_frame_equal
 
 from taipy.config.common.scope import Scope
 from taipy.config.config import Config
@@ -169,3 +171,87 @@ def test_migrate_to_new_path(self, tmp_path):
 
         assert ".data" not in dn.path
         assert os.path.exists(dn.path)
+
+    def test_get_downloadable_path(self):
+        path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.csv")
+        dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": path, "exposed_type": "pandas"})
+        assert dn._get_downloadable_path() == path
+
+    def test_get_downloadable_path_with_not_existing_file(self):
+        dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": "NOT_EXISTING.csv", "exposed_type": "pandas"})
+        assert dn._get_downloadable_path() == ""
+
+    def test_upload(self, csv_file, tmpdir_factory):
+        old_csv_path = tmpdir_factory.mktemp("data").join("df.csv").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": old_csv_path, "exposed_type": "pandas"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        upload_content = pd.read_csv(csv_file)
+        dn._upload(csv_file)
+
+        assert_frame_equal(dn.read(), upload_content)  # The content of the dn should change to the uploaded content
+        assert dn.last_edit_date > old_last_edit_date
+        assert dn.path == old_csv_path  # The path of the dn should not change
+
+    def test_upload_with_upload_check_pandas(self, csv_file, tmpdir_factory):
+        old_csv_path = tmpdir_factory.mktemp("data").join("df.csv").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": old_csv_path, "exposed_type": "pandas"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_column(upload_path, upload_data):
+            return upload_path.endswith(".csv") and upload_data.columns.tolist() == ["a", "b", "c"]
+
+        wrong_format_not_csv_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_csv").strpath
+        wrong_format_csv_path = tmpdir_factory.mktemp("data").join("wrong_format_df.csv").strpath
+        pd.DataFrame([{"a": 1, "b": 2, "d": 3}, {"a": 4, "b": 5, "d": 6}]).to_csv(wrong_format_csv_path, index=False)
+
+        # The upload should fail when the file is not a csv
+        assert not dn._upload(wrong_format_not_csv_path, upload_checker=check_data_column)
+
+        # The upload should fail when check_data_column() return False
+        assert not dn._upload(wrong_format_csv_path, upload_checker=check_data_column)
+
+        assert_frame_equal(dn.read(), old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_csv_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_column() return True
+        assert dn._upload(csv_file, upload_checker=check_data_column)
+
+    def test_upload_with_upload_check_numpy(self, tmpdir_factory):
+        old_csv_path = tmpdir_factory.mktemp("data").join("df.csv").strpath
+        old_data = np.array([[1, 2, 3], [4, 5, 6]])
+
+        new_csv_path = tmpdir_factory.mktemp("data").join("new_upload_data.csv").strpath
+        new_data = np.array([[1, 2, 3], [4, 5, 6]])
+        pd.DataFrame(new_data).to_csv(new_csv_path, index=False)
+
+        dn = CSVDataNode("foo", Scope.SCENARIO, properties={"path": old_csv_path, "exposed_type": "numpy"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_is_positive(upload_path, upload_data):
+            return upload_path.endswith(".csv") and np.all(upload_data > 0)
+
+        wrong_format_not_csv_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_csv").strpath
+        wrong_format_csv_path = tmpdir_factory.mktemp("data").join("wrong_format_df.csv").strpath
+        pd.DataFrame(np.array([[-1, 2, 3], [-4, -5, -6]])).to_csv(wrong_format_csv_path, index=False)
+
+        # The upload should fail when the file is not a csv
+        assert not dn._upload(wrong_format_not_csv_path, upload_checker=check_data_is_positive)
+
+        # The upload should fail when check_data_is_positive() return False
+        assert not dn._upload(wrong_format_csv_path, upload_checker=check_data_is_positive)
+
+        np.array_equal(dn.read(), old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_csv_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_is_positive() return True
+        assert dn._upload(new_csv_path, upload_checker=check_data_is_positive)
diff --git a/tests/core/data/test_excel_data_node.py b/tests/core/data/test_excel_data_node.py
index 4500caa221..c6ae3e652f 100644
--- a/tests/core/data/test_excel_data_node.py
+++ b/tests/core/data/test_excel_data_node.py
@@ -19,6 +19,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+from pandas.testing import assert_frame_equal
 
 from taipy.config.common.scope import Scope
 from taipy.config.config import Config
@@ -365,3 +366,90 @@ def test_migrate_to_new_path(self, tmp_path):
 
         assert ".data" not in dn.path
         assert os.path.exists(dn.path)
+
+    def test_get_download_path(self):
+        path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.xlsx")
+        dn = ExcelDataNode("foo", Scope.SCENARIO, properties={"path": path, "exposed_type": "pandas"})
+        assert dn._get_downloadable_path() == path
+
+    def test_get_downloadable_path_with_not_existing_file(self):
+        dn = ExcelDataNode("foo", Scope.SCENARIO, properties={"path": "NOT_EXISTING.xlsx", "exposed_type": "pandas"})
+        assert dn._get_downloadable_path() == ""
+
+    def test_upload(self, excel_file, tmpdir_factory):
+        old_xlsx_path = tmpdir_factory.mktemp("data").join("df.xlsx").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = ExcelDataNode("foo", Scope.SCENARIO, properties={"path": old_xlsx_path, "exposed_type": "pandas"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        upload_content = pd.read_excel(excel_file)
+        dn._upload(excel_file)
+
+        assert_frame_equal(dn.read()["Sheet1"], upload_content)  # The data of dn should change to the uploaded content
+        assert dn.last_edit_date > old_last_edit_date
+        assert dn.path == old_xlsx_path  # The path of the dn should not change
+
+    def test_upload_with_upload_check_pandas(self, excel_file, tmpdir_factory):
+        old_xlsx_path = tmpdir_factory.mktemp("data").join("df.xlsx").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = ExcelDataNode("foo", Scope.SCENARIO, properties={"path": old_xlsx_path, "exposed_type": "pandas"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_column(upload_path, upload_data):
+            """Check if the uploaded data has the correct file format and
+            the sheet named "Sheet1" has the correct columns.
+            """
+            return upload_path.endswith(".xlsx") and upload_data["Sheet1"].columns.tolist() == ["a", "b", "c"]
+
+        wrong_format_not_xlsx_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_xlsx").strpath
+        wrong_format_xlsx_path = tmpdir_factory.mktemp("data").join("wrong_format_df.xlsx").strpath
+        pd.DataFrame([{"a": 1, "b": 2, "d": 3}, {"a": 4, "b": 5, "d": 6}]).to_excel(wrong_format_xlsx_path, index=False)
+
+        # The upload should fail when the file is not a xlsx
+        assert not dn._upload(wrong_format_not_xlsx_path, upload_checker=check_data_column)
+
+        # The upload should fail when check_data_column() return False
+        assert not dn._upload(wrong_format_xlsx_path, upload_checker=check_data_column)
+
+        assert_frame_equal(dn.read()["Sheet1"], old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_xlsx_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_column() return True
+        assert dn._upload(excel_file, upload_checker=check_data_column)
+
+    def test_upload_with_upload_check_numpy(self, tmpdir_factory):
+        old_excel_path = tmpdir_factory.mktemp("data").join("df.xlsx").strpath
+        old_data = np.array([[1, 2, 3], [4, 5, 6]])
+
+        new_excel_path = tmpdir_factory.mktemp("data").join("new_upload_data.xlsx").strpath
+        new_data = np.array([[1, 2, 3], [4, 5, 6]])
+        pd.DataFrame(new_data).to_excel(new_excel_path, index=False)
+
+        dn = ExcelDataNode("foo", Scope.SCENARIO, properties={"path": old_excel_path, "exposed_type": "numpy"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_is_positive(upload_path, upload_data):
+            return upload_path.endswith(".xlsx") and np.all(upload_data["Sheet1"] > 0)
+
+        wrong_format_not_excel_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_excel").strpath
+        wrong_format_excel_path = tmpdir_factory.mktemp("data").join("wrong_format_df.xlsx").strpath
+        pd.DataFrame(np.array([[-1, 2, 3], [-4, -5, -6]])).to_excel(wrong_format_excel_path, index=False)
+
+        # The upload should fail when the file is not a excel
+        assert not dn._upload(wrong_format_not_excel_path, upload_checker=check_data_is_positive)
+
+        # The upload should fail when check_data_is_positive() return False
+        assert not dn._upload(wrong_format_excel_path, upload_checker=check_data_is_positive)
+
+        np.array_equal(dn.read()["Sheet1"], old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_excel_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_is_positive() return True
+        assert dn._upload(new_excel_path, upload_checker=check_data_is_positive)
diff --git a/tests/core/data/test_json_data_node.py b/tests/core/data/test_json_data_node.py
index cb48fc8c97..c2059ecd3f 100644
--- a/tests/core/data/test_json_data_node.py
+++ b/tests/core/data/test_json_data_node.py
@@ -370,3 +370,58 @@ def test_migrate_to_new_path(self, tmp_path):
 
         assert ".data" not in dn.path
         assert os.path.exists(dn.path)
+
+    def test_get_download_path(self):
+        path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/json/example_dict.json")
+        dn = JSONDataNode("foo", Scope.SCENARIO, properties={"path": path})
+        assert dn._get_downloadable_path() == path
+
+    def test_get_download_path_with_not_existed_file(self):
+        dn = JSONDataNode("foo", Scope.SCENARIO, properties={"path": "NOT_EXISTED.json"})
+        assert dn._get_downloadable_path() == ""
+
+    def test_upload(self, json_file, tmpdir_factory):
+        old_json_path = tmpdir_factory.mktemp("data").join("df.json").strpath
+        old_data = [{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}]
+
+        dn = JSONDataNode("foo", Scope.SCENARIO, properties={"path": old_json_path})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        with open(json_file, "r") as f:
+            upload_content = json.load(f)
+        dn._upload(json_file)
+
+        assert dn.read() == upload_content  # The content of the dn should change to the uploaded content
+        assert dn.last_edit_date > old_last_edit_date
+        assert dn.path == old_json_path  # The path of the dn should not change
+
+    def test_upload_with_upload_check(self, json_file, tmpdir_factory):
+        old_json_path = tmpdir_factory.mktemp("data").join("df.json").strpath
+        old_data = [{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}]
+
+        dn = JSONDataNode("foo", Scope.SCENARIO, properties={"path": old_json_path})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_keys(upload_path, upload_data):
+            all_column_is_abc = all(data.keys() == {"a", "b", "c"} for data in upload_data)
+            return upload_path.endswith(".json") and all_column_is_abc
+
+        wrong_format_not_json_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_json").strpath
+        wrong_format_json_path = tmpdir_factory.mktemp("data").join("wrong_format_df.json").strpath
+        with open(wrong_format_not_json_path, "w") as f:
+            json.dump([{"a": 1, "b": 2, "d": 3}, {"a": 4, "b": 5, "d": 6}], f)
+
+        # The upload should fail when the file is not a json
+        assert not dn._upload(wrong_format_not_json_path, upload_checker=check_data_keys)
+
+        # The upload should fail when check_data_keys() return False
+        assert not dn._upload(wrong_format_json_path, upload_checker=check_data_keys)
+
+        assert dn.read() == old_data  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_json_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_keys() return True
+        assert dn._upload(json_file, upload_checker=check_data_keys)
diff --git a/tests/core/data/test_parquet_data_node.py b/tests/core/data/test_parquet_data_node.py
index 064af6acd5..ba2ca30807 100644
--- a/tests/core/data/test_parquet_data_node.py
+++ b/tests/core/data/test_parquet_data_node.py
@@ -16,8 +16,10 @@
 from importlib import util
 from time import sleep
 
+import numpy as np
 import pandas as pd
 import pytest
+from pandas.testing import assert_frame_equal
 
 from taipy.config.common.scope import Scope
 from taipy.config.config import Config
@@ -221,3 +223,94 @@ def test_migrate_to_new_path(self, tmp_path):
 
         assert ".data" not in dn.path
         assert os.path.exists(dn.path)
+
+    def test_get_downloadable_path(self):
+        path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.parquet")
+        dn = ParquetDataNode("foo", Scope.SCENARIO, properties={"path": path, "exposed_type": "pandas"})
+        assert dn._get_downloadable_path() == path
+
+    def test_get_downloadable_path_with_not_existing_file(self):
+        dn = ParquetDataNode("foo", Scope.SCENARIO, properties={"path": "NOT_EXISTING.parquet"})
+        assert dn._get_downloadable_path() == ""
+
+    def test_get_downloadable_path_as_directory_should_return_nothing(self):
+        path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/parquet_example")
+        dn = ParquetDataNode("foo", Scope.SCENARIO, properties={"path": path})
+        assert dn._get_downloadable_path() == ""
+
+    def test_upload(self, parquet_file_path, tmpdir_factory):
+        old_parquet_path = tmpdir_factory.mktemp("data").join("df.parquet").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = ParquetDataNode("foo", Scope.SCENARIO, properties={"path": old_parquet_path, "exposed_type": "pandas"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        upload_content = pd.read_parquet(parquet_file_path)
+        dn._upload(parquet_file_path)
+
+        assert_frame_equal(dn.read(), upload_content)  # The content of the dn should change to the uploaded content
+        assert dn.last_edit_date > old_last_edit_date
+        assert dn.path == old_parquet_path  # The path of the dn should not change
+
+    def test_upload_with_upload_check_pandas(self, parquet_file_path, tmpdir_factory):
+        old_parquet_path = tmpdir_factory.mktemp("data").join("df.parquet").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = ParquetDataNode("foo", Scope.SCENARIO, properties={"path": old_parquet_path, "exposed_type": "pandas"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_column(upload_path, upload_data):
+            return upload_path.endswith(".parquet") and upload_data.columns.tolist() == ["a", "b", "c"]
+
+        wrong_format_not_parquet_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_parquet").strpath
+        wrong_format_parquet_path = tmpdir_factory.mktemp("data").join("wrong_format_df.parquet").strpath
+        pd.DataFrame([{"a": 1, "b": 2, "d": 3}, {"a": 4, "b": 5, "d": 6}]).to_parquet(
+            wrong_format_parquet_path, index=False
+        )
+
+        # The upload should fail when the file is not a parquet
+        assert not dn._upload(wrong_format_not_parquet_path, upload_checker=check_data_column)
+
+        # The upload should fail when check_data_column() return False
+        assert not dn._upload(wrong_format_parquet_path, upload_checker=check_data_column)
+
+        assert_frame_equal(dn.read(), old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_parquet_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_column() return True
+        assert dn._upload(parquet_file_path, upload_checker=check_data_column)
+
+    def test_upload_with_upload_check_numpy(self, tmpdir_factory):
+        old_parquet_path = tmpdir_factory.mktemp("data").join("df.parquet").strpath
+        old_data = np.array([[1, 2, 3], [4, 5, 6]])
+
+        new_parquet_path = tmpdir_factory.mktemp("data").join("new_upload_data.parquet").strpath
+        new_data = np.array([[1, 2, 3], [4, 5, 6]])
+        pd.DataFrame(new_data).to_parquet(new_parquet_path, index=False)
+
+        dn = ParquetDataNode("foo", Scope.SCENARIO, properties={"path": old_parquet_path, "exposed_type": "numpy"})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_is_positive(upload_path, upload_data):
+            return upload_path.endswith(".parquet") and np.all(upload_data > 0)
+
+        wrong_format_not_parquet_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_parquet").strpath
+        wrong_format_parquet_path = tmpdir_factory.mktemp("data").join("wrong_format_df.parquet").strpath
+        pd.DataFrame(np.array([[-1, 2, 3], [-4, -5, -6]])).to_parquet(wrong_format_parquet_path, index=False)
+
+        # The upload should fail when the file is not a parquet
+        assert not dn._upload(wrong_format_not_parquet_path, upload_checker=check_data_is_positive)
+
+        # The upload should fail when check_data_is_positive() return False
+        assert not dn._upload(wrong_format_parquet_path, upload_checker=check_data_is_positive)
+
+        np.array_equal(dn.read(), old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_parquet_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_is_positive() return True
+        assert dn._upload(new_parquet_path, upload_checker=check_data_is_positive)
diff --git a/tests/core/data/test_pickle_data_node.py b/tests/core/data/test_pickle_data_node.py
index 3cdb33ef7e..34b5123e64 100644
--- a/tests/core/data/test_pickle_data_node.py
+++ b/tests/core/data/test_pickle_data_node.py
@@ -11,11 +11,13 @@
 
 import os
 import pathlib
+import pickle
 from datetime import datetime
 from time import sleep
 
 import pandas as pd
 import pytest
+from pandas.testing import assert_frame_equal
 
 from taipy.config.common.scope import Scope
 from taipy.config.config import Config
@@ -192,3 +194,56 @@ def test_migrate_to_new_path(self, tmp_path):
 
         assert ".data" not in dn.path
         assert os.path.exists(dn.path)
+
+    def test_get_download_path(self):
+        path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/example.p")
+        dn = PickleDataNode("foo", Scope.SCENARIO, properties={"path": path})
+        assert dn._get_downloadable_path() == path
+
+    def test_get_download_path_with_not_existed_file(self):
+        dn = PickleDataNode("foo", Scope.SCENARIO, properties={"path": "NOT_EXISTED.p"})
+        assert dn._get_downloadable_path() == ""
+
+    def test_upload(self, pickle_file_path, tmpdir_factory):
+        old_pickle_path = tmpdir_factory.mktemp("data").join("df.p").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = PickleDataNode("foo", Scope.SCENARIO, properties={"path": old_pickle_path})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        upload_content = pd.read_pickle(pickle_file_path)
+        dn._upload(pickle_file_path)
+
+        assert_frame_equal(dn.read(), upload_content)  # The content of the dn should change to the uploaded content
+        assert dn.last_edit_date > old_last_edit_date
+        assert dn.path == old_pickle_path  # The path of the dn should not change
+
+    def test_upload_with_upload_check(self, pickle_file_path, tmpdir_factory):
+        old_pickle_path = tmpdir_factory.mktemp("data").join("df.p").strpath
+        old_data = pd.DataFrame([{"a": 0, "b": 1, "c": 2}, {"a": 3, "b": 4, "c": 5}])
+
+        dn = PickleDataNode("foo", Scope.SCENARIO, properties={"path": old_pickle_path})
+        dn.write(old_data)
+        old_last_edit_date = dn.last_edit_date
+
+        def check_data_column(upload_path, upload_data):
+            return upload_path.endswith(".p") and upload_data.columns.tolist() == ["a", "b", "c"]
+
+        wrong_format_not_pickle_path = tmpdir_factory.mktemp("data").join("wrong_format_df.not_pickle").strpath
+        wrong_format_pickle_path = tmpdir_factory.mktemp("data").join("wrong_format_df.p").strpath
+        with open(str(wrong_format_pickle_path), "wb") as f:
+            pickle.dump(pd.DataFrame([{"a": 1, "b": 2, "d": 3}, {"a": 4, "b": 5, "d": 6}]), f)
+
+        # The upload should fail when the file is not a pickle
+        assert not dn._upload(wrong_format_not_pickle_path, upload_checker=check_data_column)
+
+        # The upload should fail when check_data_column() return False
+        assert not dn._upload(wrong_format_pickle_path, upload_checker=check_data_column)
+
+        assert_frame_equal(dn.read(), old_data)  # The content of the dn should not change when upload fails
+        assert dn.last_edit_date == old_last_edit_date  # The last edit date should not change when upload fails
+        assert dn.path == old_pickle_path  # The path of the dn should not change
+
+        # The upload should succeed when check_data_column() return True
+        assert dn._upload(pickle_file_path, upload_checker=check_data_column)


Code review comments:
diff_hunk: @@ -27,7 +27,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class CSVDataNode(DataNode, _FileDataNodeMixin, _TabularDataNodeMixin):
+class CSVDataNode(_FileDataNodeMixin, DataNode, _TabularDataNodeMixin):
path: taipy/core/data/csv.py
text: @user1:
hmm? Don't we initiated them all manually? so the order shouldn't be an issue? 

@author:
It does in this case.
So we have `_upload()` in both `DataNode` and `_FileDataNodeMixin`.
If we put DataNode first, when calling `CSVDataNode._upload()`, it will always go to `DataNode._upload()`, which is not what we want here.

diff_hunk: @@ -92,3 +92,50 @@ def _migrate_path(self, storage_type, old_path) -> str:
         if os.path.exists(old_path):
             shutil.move(old_path, new_path)
         return new_path
+
+    def _get_downloadable_path(self) -> str:
+        """Get the downloadable path of the file data of the data node.
+
+        Returns:
+            The downloadable path of the file data of the data node if it exists, otherwise an empty string.
+        """
+        if os.path.exists(self.path) and isfile(self._path):
+            return self.path
+
+        return ""
+
+    def _upload(self, path: str, upload_checker: Optional[Callable[[str, Any], bool]] = None) -> bool:
+        """Upload a csv file data to the data node.
+
+        Parameters:
+            path (str): The path of the file to upload to the data node.
+            upload_checker (Optional[Callable[[str, Any], bool]]): A function to check if the upload is allowed.
+                The function takes the title of the upload data and the data itself as arguments and returns
+                True if the upload is allowed, otherwise False.
+
+        Returns:
+            True if the upload was successful, otherwise False.
+        """
+        from ._data_manager_factory import _DataManagerFactory
+
+        upload_path = pathlib.Path(path)
+
+        if upload_checker is not None:
+            try:
+                upload_data = self._read_from_path(upload_path.name)
+            except Exception:
+                return False
path: taipy/core/data/_file_datanode_mixin.py
text: @author:
For now, we only return a bool if there is any Exception.

Should we instead:
- raise the Exception 
- log the error and return False

@user1:
Yes, probably log the error and raise a custom exception.

diff_hunk: @@ -92,3 +92,50 @@ def _migrate_path(self, storage_type, old_path) -> str:
         if os.path.exists(old_path):
             shutil.move(old_path, new_path)
         return new_path
+
+    def _get_downloadable_path(self) -> str:
+        """Get the downloadable path of the file data of the data node.
+
+        Returns:
+            The downloadable path of the file data of the data node if it exists, otherwise an empty string.
+        """
+        if os.path.exists(self.path) and isfile(self._path):
+            return self.path
+
+        return ""
+
+    def _upload(self, path: str, upload_checker: Optional[Callable[[str, Any], bool]] = None) -> bool:
+        """Upload a csv file data to the data node.
+
+        Parameters:
+            path (str): The path of the file to upload to the data node.
+            upload_checker (Optional[Callable[[str, Any], bool]]): A function to check if the upload is allowed.
+                The function takes the title of the upload data and the data itself as arguments and returns
+                True if the upload is allowed, otherwise False.
+
+        Returns:
+            True if the upload was successful, otherwise False.
+        """
+        from ._data_manager_factory import _DataManagerFactory
+
+        upload_path = pathlib.Path(path)
+
+        if upload_checker is not None:
+            try:
+                upload_data = self._read_from_path(upload_path.name)
+            except Exception:
+                return False
+
+            if not upload_checker(upload_path.name, upload_data):
+                return False
path: taipy/core/data/_file_datanode_mixin.py
text: @author:
For now, we only return a bool if the checker return False. Should we return a `Reasons`?

@user1:
I think we should :D

@user2:
I would say yes.

diff_hunk: @@ -27,7 +27,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class CSVDataNode(DataNode, _FileDataNodeMixin, _TabularDataNodeMixin):
+class CSVDataNode(_FileDataNodeMixin, DataNode, _TabularDataNodeMixin):
path: taipy/core/data/csv.py
text: @user1:
WHy don't you want to extend DataNode first?

diff_hunk: @@ -92,3 +92,50 @@ def _migrate_path(self, storage_type, old_path) -> str:
         if os.path.exists(old_path):
             shutil.move(old_path, new_path)
         return new_path
+
+    def _get_downloadable_path(self) -> str:
+        """Get the downloadable path of the file data of the data node.
+
+        Returns:
+            The downloadable path of the file data of the data node if it exists, otherwise an empty string.
+        """
+        if os.path.exists(self.path) and isfile(self._path):
+            return self.path
+
+        return ""
+
+    def _upload(self, path: str, upload_checker: Optional[Callable[[str, Any], bool]] = None) -> bool:
+        """Upload a csv file data to the data node.
path: taipy/core/data/_file_datanode_mixin.py
text: @user1:
```suggestion
        """Upload a file data to the data node.
```

diff_hunk: @@ -373,6 +373,28 @@ def _get_last_modified_datetime(cls, path: Optional[str] = None) -> Optional[dat
     def storage_type(cls) -> str:
         raise NotImplementedError
 
+    def _get_downloadable_path(self) -> str:
path: taipy/core/data/data_node.py
text: @user1:
this has already defined in the FileDataNodeMixin no?

@author:
Yes. The idea is that we want the method to only defined in `_ FileDataNodeMixin`.

`SQLDataNode. _get_downloadable_path()` should raise the NotImplementedError error.

@user1:
ah ok, then I would suggest to move this code in to _abstract_sql.py instead, since DataNode and _FileDataNodeMixin are 2 different abstract class, I don't think they will have much connection in this sense.

@author:
No I think I misrepresented my idea.

So, for all data nodes that are not File-based, calling to `_get_downloadable_path` should raise an error.
For file-based dn, it should work.

@user1:
hmm in that case, I think it should still be fine if you put it in the _abstract_sql, since other DataNode are not extending from `_FileDataNodeMixin` so it should be fine for them no? 

@author:
what about other classes like Mongo, S3, Generic, ...?
If we do that, we also need to add one for each of them, which is weird.

@user1:
hmmm I don't think they are extending the class `_FileDataNodeMixin` :/

diff_hunk: @@ -28,7 +28,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class ParquetDataNode(DataNode, _FileDataNodeMixin, _TabularDataNodeMixin):
+class ParquetDataNode(_FileDataNodeMixin, DataNode, _TabularDataNodeMixin):
path: taipy/core/data/parquet.py
text: @user1:
Same here abt the order. I prefer to inherit from the data node first.

diff_hunk: @@ -373,6 +373,28 @@ def _get_last_modified_datetime(cls, path: Optional[str] = None) -> Optional[dat
     def storage_type(cls) -> str:
         raise NotImplementedError
 
+    def _get_downloadable_path(self) -> str:
+        """Get the downloadable path of the file data of the data node.
+
+        Returns:
+            The downloadable path of the file data of the data node.
+        """
+        raise NotImplementedError
+
+    def _upload(self, path: str, upload_checker: Optional[Callable[[str, Any], bool]] = None) -> bool:
+        """Upload the file data of the data node.
+
+        Parameters:
+            path (str): The path of the file to upload to the data node.
+            upload_checker (Optional[Callable[[str, Any], bool]]): A function to check if the upload is allowed.
+                The function takes the title of the upload data and the data itself as arguments and returns
+                True if the upload is allowed, otherwise False.
+
+        Returns:
+            True if the upload was successful, otherwise False.
+        """
+        raise NotImplementedError
path: taipy/core/data/data_node.py
text: @user1:
We don't want these to be part of a DataNode. They are only specific to the file data nodes, no?
No need to define the methods here. and so in each data node class (csv, excel, json, etc.) you can inherit from data node first.

diff_hunk: @@ -25,7 +25,7 @@
 from .data_node_id import DataNodeId, Edit
 
 
-class JSONDataNode(DataNode, _FileDataNodeMixin):
+class JSONDataNode(_FileDataNodeMixin, DataNode):
path: taipy/core/data/json.py
text: @user1:
Same here, I would prefer to inherit from data node first.