Repository: Avaiga/taipy
Pull request: Feature/#410 - Import scenario from an exported one
Resolves #410 

Problem statement:
Possbility to import a scenario (after an export)
**What would that feature address**
The possibility of importing a scenario/scenarios after exporting it with Taipy. The problem is that in the context of a project, different scenarios are present in different environments (Test, QA, Production). 

For example, the end user tests its application on the Test environment and creates meaningful scenarios. When the test is done, the application goes into QA and Production. The end user wants the scenarios he created in the other environments.

Another use case could be if the end-user encounters an issue with a scenario/scenarios he has created; he would then want to send it/them to its technical team (Data Scientist, Python Developers, ...) to debug them.

***Description of the ideal solution***
This export/import system should work with the different storage methods (Filesystem, SQL, Mongo). It could be another option of the CLI.
The goal would be to have the possibility to import a scenario. The difficulty comes from ensuring the consistency of the import regarding the entities related to the scenario and the data attached to data nodes.


**Acceptance Criteria**
- [ ] Ensure new code is unit tested, and check code coverage is at least 90%
- [ ] Create related issue in taipy-doc for documentation and Release Notes
- [ ] Check if a new demo could be provided based on this, or if legacy demos could be benefit from it
- [ ] Ensure any change is well documented

Hints from issue discussion:
For the current implementation of the `taipy.export()` API, user data is not exported alongside the entities.

We will need to check if the data is Taipy generated, we should also export the user data.

The new signature of the export method can be:
```
taipy.export(include_data=True)
```

When including the data, we will only export the file-based data node's data.

Commit message: fix: mypy errors

Patch to review:
diff --git a/taipy/core/_manager/_manager.py b/taipy/core/_manager/_manager.py
index 7219cec97f..bf406b1bdb 100644
--- a/taipy/core/_manager/_manager.py
+++ b/taipy/core/_manager/_manager.py
@@ -157,6 +157,13 @@ def _delete_entities_of_multiple_types(cls, _entity_ids: _EntityIds):
     def _export(cls, id: str, folder_path: Union[str, pathlib.Path], **kwargs):
         return cls._repository._export(id, folder_path)
 
+    @classmethod
+    def _import(cls, entity_file: pathlib.Path, version: str, **kwargs) -> EntityType:
+        imported_entity = cls._repository._import(entity_file)
+        imported_entity._version = version
+        cls._set(imported_entity)
+        return imported_entity
+
     @classmethod
     def _is_editable(cls, entity: Union[EntityType, str]) -> bool:
         return True
diff --git a/taipy/core/_repository/_abstract_repository.py b/taipy/core/_repository/_abstract_repository.py
index 6f810f1420..31319fc283 100644
--- a/taipy/core/_repository/_abstract_repository.py
+++ b/taipy/core/_repository/_abstract_repository.py
@@ -9,10 +9,14 @@
 # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations under the License.
 
+import json
 import pathlib
 from abc import abstractmethod
 from typing import Any, Dict, Generic, Iterable, List, Optional, TypeVar, Union
 
+from ..exceptions import FileCannotBeRead
+from ._decoder import _Decoder
+
 ModelType = TypeVar("ModelType")
 Entity = TypeVar("Entity")
 
@@ -122,3 +126,27 @@ def _export(self, entity_id: str, folder_path: Union[str, pathlib.Path]):
             folder_path (Union[str, pathlib.Path]): The folder path to export the entity to.
         """
         raise NotImplementedError
+
+    def _import(self, entity_file_path: pathlib.Path) -> Entity:
+        """
+        Import an entity from an exported file.
+
+        Parameters:
+            folder_path (Union[str, pathlib.Path]): The folder path to export the entity to.
+
+        Returns:
+            The imported entity.
+        """
+        if not entity_file_path.is_file():
+            raise FileNotFoundError
+
+        try:
+            with entity_file_path.open("r", encoding="UTF-8") as f:
+                file_content = f.read()
+        except Exception:
+            raise FileCannotBeRead(str(entity_file_path)) from None
+
+        if isinstance(file_content, str):
+            file_content = json.loads(file_content, cls=_Decoder)
+        model = self.model_type.from_dict(file_content)  # type: ignore[attr-defined]
+        return self.converter._model_to_entity(model)  # type: ignore[attr-defined]
diff --git a/taipy/core/_version/_version_manager.py b/taipy/core/_version/_version_manager.py
index 24863888a7..6ff5d0fa80 100644
--- a/taipy/core/_version/_version_manager.py
+++ b/taipy/core/_version/_version_manager.py
@@ -9,6 +9,7 @@
 # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations under the License.
 
+import pathlib
 import uuid
 from typing import List, Optional, Union
 
@@ -230,3 +231,17 @@ def __check_production_migration_config(cls):
     @classmethod
     def _delete_entities_of_multiple_types(cls, _entity_ids):
         raise NotImplementedError
+
+    @classmethod
+    def _import(cls, entity_file: pathlib.Path, version: str, **kwargs) -> _Version:
+        imported_version = cls._repository._import(entity_file)
+
+        comparator_result = Config._comparator._find_conflict_config(  # type: ignore[attr-defined]
+            imported_version.config,
+            Config._applied_config,  # type: ignore[attr-defined]
+            imported_version.id,
+        )
+        if comparator_result.get(_ComparatorResult.CONFLICTED_SECTION_KEY):
+            raise ConflictedConfigurationError()
+
+        return imported_version
diff --git a/taipy/core/_version/_version_manager_factory.py b/taipy/core/_version/_version_manager_factory.py
index 2ec3d9baac..5818afca46 100644
--- a/taipy/core/_version/_version_manager_factory.py
+++ b/taipy/core/_version/_version_manager_factory.py
@@ -9,6 +9,8 @@
 # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations under the License.
 
+from typing import Type
+
 from .._manager._manager_factory import _ManagerFactory
 from ..common import _utils
 from ._version_fs_repository import _VersionFSRepository
@@ -17,11 +19,10 @@
 
 
 class _VersionManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _VersionFSRepository, "sql": _VersionSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> _VersionManager:  # type: ignore
+    def _build_manager(cls) -> Type[_VersionManager]:
         if cls._using_enterprise():
             version_manager = _utils._load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + "._version._version_manager", "_VersionManager"
diff --git a/taipy/core/cycle/_cycle_manager_factory.py b/taipy/core/cycle/_cycle_manager_factory.py
index 579fc5ae3c..b7506a4e6e 100644
--- a/taipy/core/cycle/_cycle_manager_factory.py
+++ b/taipy/core/cycle/_cycle_manager_factory.py
@@ -19,11 +19,10 @@
 
 
 class _CycleManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _CycleFSRepository, "sql": _CycleSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> Type[_CycleManager]:  # type: ignore
+    def _build_manager(cls) -> Type[_CycleManager]:
         if cls._using_enterprise():
             cycle_manager = _load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + ".cycle._cycle_manager", "_CycleManager"
diff --git a/taipy/core/data/_data_manager.py b/taipy/core/data/_data_manager.py
index a514b692c7..cc06c0e7a9 100644
--- a/taipy/core/data/_data_manager.py
+++ b/taipy/core/data/_data_manager.py
@@ -181,10 +181,28 @@ def _export(cls, id: str, folder_path: Union[str, pathlib.Path], **kwargs):
         else:
             folder = folder_path
 
-        data_export_dir = folder / Config.core.storage_folder
+        data_export_dir = folder / Config.core.storage_folder / os.path.dirname(data_node.path)
         if not data_export_dir.exists():
             data_export_dir.mkdir(parents=True)
 
         data_export_path = data_export_dir / os.path.basename(data_node.path)
         if os.path.exists(data_node.path):
-            shutil.copy(data_node.path, data_export_path)
+            shutil.copy2(data_node.path, data_export_path)
+
+    @classmethod
+    def _import(cls, entity_file: pathlib.Path, version: str, **kwargs) -> DataNode:
+        imported_data_node = cls._repository._import(entity_file)
+        imported_data_node._version = version
+        cls._set(imported_data_node)
+
+        if not (isinstance(imported_data_node, _FileDataNodeMixin) and isinstance(imported_data_node, DataNode)):
+            return imported_data_node
+
+        data_folder: pathlib.Path = pathlib.Path(str(kwargs.get("data_folder")))
+        if not data_folder.exists():
+            return imported_data_node
+
+        if (data_folder / imported_data_node.path).exists():
+            shutil.copy2(data_folder / imported_data_node.path, imported_data_node.path)
+
+        return imported_data_node
diff --git a/taipy/core/data/_data_manager_factory.py b/taipy/core/data/_data_manager_factory.py
index 4ef3358e8d..bb165279cc 100644
--- a/taipy/core/data/_data_manager_factory.py
+++ b/taipy/core/data/_data_manager_factory.py
@@ -19,11 +19,10 @@
 
 
 class _DataManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _DataFSRepository, "sql": _DataSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> Type[_DataManager]:  # type: ignore
+    def _build_manager(cls) -> Type[_DataManager]:
         if cls._using_enterprise():
             data_manager = _load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + ".data._data_manager", "_DataManager"
diff --git a/taipy/core/exceptions/exceptions.py b/taipy/core/exceptions/exceptions.py
index c6b7f57486..c8c886060b 100644
--- a/taipy/core/exceptions/exceptions.py
+++ b/taipy/core/exceptions/exceptions.py
@@ -261,7 +261,7 @@ def __init__(self, scenario_config_id: str):
         self.message = f"Scenario config: {scenario_config_id} does not exist."
 
 
-class InvalidSscenario(Exception):
+class InvalidScenario(Exception):
     """Raised if a Scenario is not a Directed Acyclic Graph."""
 
     def __init__(self, scenario_id: str):
@@ -339,10 +339,6 @@ class ModeNotAvailable(Exception):
     """Raised if the mode in JobConfig is not supported."""
 
 
-class InvalidExportPath(Exception):
-    """Raised if the export path is not valid."""
-
-
 class NonExistingVersion(Exception):
     """Raised if request a Version that is not known by the Version Manager."""
 
@@ -373,16 +369,47 @@ class FileCannotBeRead(Exception):
     """Raised when a file cannot be read."""
 
 
-class ExportFolderAlreadyExists(Exception):
+class ExportPathAlreadyExists(Exception):
     """Raised when the export folder already exists."""
 
-    def __init__(self, folder_path: str, scenario_id: str):
+    def __init__(self, export_path: str, scenario_id: str):
         self.message = (
-            f"Folder '{folder_path}' already exists and can not be used to export scenario '{scenario_id}'."
+            f"The path '{export_path}' already exists and can not be used to export scenario '{scenario_id}'."
             " Please use the 'override' parameter to override it."
         )
 
 
+class EntitiesToBeImportAlredyExist(Exception):
+    """Raised when entities in the scenario to be imported have already exists"""
+
+    def __init__(self, import_path):
+        self.message = f"The import archive file {import_path} contains entities that have already existed."
+
+
+class DataToBeImportAlredyExist(Exception):
+    """Raised when data files in the scenario to be imported have already exists"""
+
+    def __init__(self, import_path):
+        self.message = (
+            f"The import archive file {import_path} contains data files that have already existed."
+            " Please use the 'override' parameter to override those."
+        )
+
+
+class ImportArchiveDoesntContainAnyScenario(Exception):
+    """Raised when the import archive file doesn't contain any scenario"""
+
+    def __init__(self, import_path):
+        self.message = f"The import archive file {import_path} doesn't contain any scenario."
+
+
+class ImportScenarioDoesntHaveAVersion(Exception):
+    """Raised when the import scenario doesn't have a version"""
+
+    def __init__(self, import_path):
+        self.message = f"The import scenario in the import archive file {import_path} doesn't have a version."
+
+
 class SQLQueryCannotBeExecuted(Exception):
     """Raised when an SQL Query cannot be executed."""
 
diff --git a/taipy/core/job/_job_manager.py b/taipy/core/job/_job_manager.py
index 39a732ef42..c6e4f1f80b 100644
--- a/taipy/core/job/_job_manager.py
+++ b/taipy/core/job/_job_manager.py
@@ -58,7 +58,9 @@ def _create(
         return job
 
     @classmethod
-    def _delete(cls, job: Job, force=False):
+    def _delete(cls, job: Union[Job, JobId], force=False):
+        if isinstance(job, str):
+            job = cls._get(job)
         if cls._is_deletable(job) or force:
             super()._delete(job.id)
         else:
diff --git a/taipy/core/job/_job_manager_factory.py b/taipy/core/job/_job_manager_factory.py
index b6d5f15a18..708a6b79d9 100644
--- a/taipy/core/job/_job_manager_factory.py
+++ b/taipy/core/job/_job_manager_factory.py
@@ -19,11 +19,10 @@
 
 
 class _JobManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _JobFSRepository, "sql": _JobSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> Type[_JobManager]:  # type: ignore
+    def _build_manager(cls) -> Type[_JobManager]:
         if cls._using_enterprise():
             job_manager = _load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + ".job._job_manager", "_JobManager"
diff --git a/taipy/core/scenario/_scenario_manager.py b/taipy/core/scenario/_scenario_manager.py
index 8b83b1b6ea..d31b680106 100644
--- a/taipy/core/scenario/_scenario_manager.py
+++ b/taipy/core/scenario/_scenario_manager.py
@@ -10,14 +10,18 @@
 # specific language governing permissions and limitations under the License.
 
 import datetime
+import pathlib
+import tempfile
+import zipfile
 from functools import partial
-from typing import Any, Callable, List, Literal, Optional, Union
+from typing import Any, Callable, Dict, List, Literal, Optional, Type, Union
 
 from taipy.config import Config
 
 from .._entity._entity_ids import _EntityIds
 from .._manager._manager import _Manager
 from .._repository._abstract_repository import _AbstractRepository
+from .._version._version_manager_factory import _VersionManagerFactory
 from .._version._version_mixin import _VersionMixin
 from ..common.warn_if_inputs_not_ready import _warn_if_inputs_not_ready
 from ..config.scenario_config import ScenarioConfig
@@ -28,9 +32,12 @@
     DeletingPrimaryScenario,
     DifferentScenarioConfigs,
     DoesNotBelongToACycle,
+    EntitiesToBeImportAlredyExist,
+    ImportArchiveDoesntContainAnyScenario,
+    ImportScenarioDoesntHaveAVersion,
     InsufficientScenarioToCompare,
+    InvalidScenario,
     InvalidSequence,
-    InvalidSscenario,
     NonExistingComparator,
     NonExistingScenario,
     NonExistingScenarioConfig,
@@ -180,7 +187,7 @@ def _create(
         cls._set(scenario)
 
         if not scenario._is_consistent():
-            raise InvalidSscenario(scenario.id)
+            raise InvalidScenario(scenario.id)
 
         actual_sequences = scenario._get_sequences()
         for sequence_name in sequences.keys():
@@ -450,3 +457,85 @@ def _get_by_config_id(cls, config_id: str, version_number: Optional[str] = None)
         for fil in filters:
             fil.update({"config_id": config_id})
         return cls._repository._load_all(filters)
+
+    @classmethod
+    def _import_scenario_and_children_entities(
+        cls,
+        zip_file_path: pathlib.Path,
+        override: bool,
+        entity_managers: Dict[str, Type[_Manager]],
+    ) -> Optional[Scenario]:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            with zipfile.ZipFile(zip_file_path) as zip_file:
+                zip_file.extractall(tmp_dir)
+
+            tmp_dir_path = pathlib.Path(tmp_dir)
+
+            if not ((tmp_dir_path / "scenarios").exists() or (tmp_dir_path / "scenario").exists()):
+                raise ImportArchiveDoesntContainAnyScenario(zip_file_path)
+
+            if not (tmp_dir_path / "version").exists():
+                raise ImportScenarioDoesntHaveAVersion(zip_file_path)
+
+            # Import the version to check for compatibility
+            entity_managers["version"]._import(next((tmp_dir_path / "version").iterdir()), "")
+
+            valid_entity_folders = list(entity_managers.keys())
+            valid_data_folder = Config.core.storage_folder
+
+            imported_scenario = None
+            imported_entities: Dict[str, List] = {}
+
+            for entity_folder in tmp_dir_path.iterdir():
+                if not entity_folder.is_dir() or entity_folder.name not in valid_entity_folders + [valid_data_folder]:
+                    cls._logger.warning(f"{entity_folder} is not a valid Taipy folder and will not be imported.")
+                    continue
+
+            try:
+                for entity_type in valid_entity_folders:
+                    # Skip the version folder as it is already handled
+                    if entity_type == "version":
+                        continue
+
+                    entity_folder = tmp_dir_path / entity_type
+                    if not entity_folder.exists():
+                        continue
+
+                    manager = entity_managers[entity_type]
+                    imported_entities[entity_type] = []
+
+                    for entity_file in entity_folder.iterdir():
+                        # Check if the to-be-imported entity already exists
+                        entity_id = entity_file.stem
+                        if manager._exists(entity_id):
+                            if override:
+                                cls._logger.warning(f"{entity_id} already exists and will be overridden.")
+                            else:
+                                cls._logger.error(
+                                    f"{entity_id} already exists. Please use the 'override' parameter to override it."
+                                )
+                                raise EntitiesToBeImportAlredyExist(zip_file_path)
+
+                        # Import the entity
+                        imported_entity = manager._import(
+                            entity_file,
+                            version=_VersionManagerFactory._build_manager()._get_latest_version(),
+                            data_folder=tmp_dir_path / valid_data_folder,
+                        )
+
+                        imported_entities[entity_type].append(imported_entity.id)
+                        if entity_type in ["scenario", "scenarios"]:
+                            imported_scenario = imported_entity
+            except Exception as err:
+                cls._logger.error(f"An error occurred during the import: {err}. Rollback the import.")
+
+                # Rollback the import
+                for entity_type, entity_ids in list(imported_entities.items())[::-1]:
+                    manager = entity_managers[entity_type]
+                    for entity_id in entity_ids:
+                        if manager._exists(entity_id):
+                            manager._delete(entity_id)
+                return None
+
+        cls._logger.info(f"Scenario {imported_scenario.id} has been successfully imported.")  # type: ignore[union-attr]
+        return imported_scenario
diff --git a/taipy/core/scenario/_scenario_manager_factory.py b/taipy/core/scenario/_scenario_manager_factory.py
index bfe0b68414..fc9169b789 100644
--- a/taipy/core/scenario/_scenario_manager_factory.py
+++ b/taipy/core/scenario/_scenario_manager_factory.py
@@ -19,11 +19,10 @@
 
 
 class _ScenarioManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _ScenarioFSRepository, "sql": _ScenarioSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> Type[_ScenarioManager]:  # type: ignore
+    def _build_manager(cls) -> Type[_ScenarioManager]:
         if cls._using_enterprise():
             scenario_manager = _load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + ".scenario._scenario_manager", "_ScenarioManager"
diff --git a/taipy/core/submission/_submission_manager_factory.py b/taipy/core/submission/_submission_manager_factory.py
index ea8aa42d3e..0b831bffc0 100644
--- a/taipy/core/submission/_submission_manager_factory.py
+++ b/taipy/core/submission/_submission_manager_factory.py
@@ -19,11 +19,10 @@
 
 
 class _SubmissionManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _SubmissionFSRepository, "sql": _SubmissionSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> Type[_SubmissionManager]:  # type: ignore
+    def _build_manager(cls) -> Type[_SubmissionManager]:
         if cls._using_enterprise():
             submission_manager = _load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + ".submission._submission_manager", "_SubmissionManager"
diff --git a/taipy/core/taipy.py b/taipy/core/taipy.py
index 8d85641138..5243a57186 100644
--- a/taipy/core/taipy.py
+++ b/taipy/core/taipy.py
@@ -12,14 +12,16 @@
 import os
 import pathlib
 import shutil
+import tempfile
 from datetime import datetime
-from typing import Any, Callable, Dict, List, Literal, Optional, Set, Union, overload
+from typing import Any, Callable, Dict, List, Literal, Optional, Set, Type, Union, overload
 
-from taipy.config import Config, Scope
+from taipy.config import Scope
 from taipy.logger._taipy_logger import _TaipyLogger
 
 from ._core import Core
 from ._entity._entity import _Entity
+from ._manager._manager import _Manager
 from ._version._version_manager_factory import _VersionManagerFactory
 from .common._check_instance import (
     _is_cycle,
@@ -41,8 +43,7 @@
 from .data.data_node_id import DataNodeId
 from .exceptions.exceptions import (
     DataNodeConfigIsNotGlobal,
-    ExportFolderAlreadyExists,
-    InvalidExportPath,
+    ExportPathAlreadyExists,
     ModelNotFound,
     NonExistingVersion,
     VersionIsNotProductionVersion,
@@ -65,7 +66,7 @@
 __logger = _TaipyLogger._get_logger()
 
 
-def set(entity: Union[DataNode, Task, Sequence, Scenario, Cycle]):
+def set(entity: Union[DataNode, Task, Sequence, Scenario, Cycle, Submission]):
     """Save or update an entity.
 
     This function allows you to save or update an entity in Taipy.
@@ -595,7 +596,6 @@ def get_primary_scenarios(
     return scenarios
 
 
-
 def is_promotable(scenario: Union[Scenario, ScenarioId]) -> bool:
     """Determine if a scenario can be promoted to become a primary scenario.
 
@@ -981,18 +981,19 @@ def clean_all_entities(version_number: str) -> bool:
 
 def export_scenario(
     scenario_id: ScenarioId,
-    folder_path: Union[str, pathlib.Path],
+    output_path: Union[str, pathlib.Path],
     override: bool = False,
     include_data: bool = False,
 ):
-    """Export all related entities of a scenario to a folder.
+    """Export all related entities of a scenario to a archive zip file.
 
     This function exports all related entities of the specified scenario to the
-    specified folder.
+    specified archive zip file.
 
     Parameters:
         scenario_id (ScenarioId): The ID of the scenario to export.
-        folder_path (Union[str, pathlib.Path]): The folder path to export the scenario to.
+        output_path (Union[str, pathlib.Path]): The path to export the scenario to.
+            The path should include the file name without the extension or with the `.zip` extension.
             If the path exists and the override parameter is False, an exception is raised.
         override (bool): If True, the existing folder will be overridden. Default is False.
         include_data (bool): If True, the file-based data nodes are exported as well.
@@ -1001,7 +1002,7 @@ def export_scenario(
             will not be exported. The default value is False.
 
     Raises:
-        ExportFolderAlreadyExist^: If the `folder_path` already exists and the override parameter is False.
+        ExportPathAlreadyExists^: If the `output_path` already exists and the override parameter is False.
     """
     manager = _ScenarioManagerFactory._build_manager()
     scenario = manager._get(scenario_id)
@@ -1010,31 +1011,82 @@ def export_scenario(
     if scenario.cycle:
         entity_ids.cycle_ids = {scenario.cycle.id}
 
-    if folder_path == Config.core.taipy_storage_folder:
-        raise InvalidExportPath("The export folder must not be the storage folder.")
+    output_filename = os.path.splitext(output_path)[0] if str(output_path).endswith(".zip") else str(output_path)
+    output_zip_path = pathlib.Path(output_filename + ".zip")
 
-    if os.path.exists(folder_path):
+    if output_zip_path.exists():
         if override:
-            __logger.warning(f"Override the existing folder '{folder_path}'")
-            shutil.rmtree(folder_path, ignore_errors=True)
+            __logger.warning(f"Override the existing path '{output_zip_path}' to export scenario {scenario_id}.")
+            output_zip_path.unlink()
         else:
-            raise ExportFolderAlreadyExists(str(folder_path), scenario_id)
-
-    for data_node_id in entity_ids.data_node_ids:
-        _DataManagerFactory._build_manager()._export(data_node_id, folder_path, include_data=include_data)
-    for task_id in entity_ids.task_ids:
-        _TaskManagerFactory._build_manager()._export(task_id, folder_path)
-    for sequence_id in entity_ids.sequence_ids:
-        _SequenceManagerFactory._build_manager()._export(sequence_id, folder_path)
-    for cycle_id in entity_ids.cycle_ids:
-        _CycleManagerFactory._build_manager()._export(cycle_id, folder_path)
-    for scenario_id in entity_ids.scenario_ids:
-        _ScenarioManagerFactory._build_manager()._export(scenario_id, folder_path)
-    for job_id in entity_ids.job_ids:
-        _JobManagerFactory._build_manager()._export(job_id, folder_path)
-    for submission_id in entity_ids.submission_ids:
-        _SubmissionManagerFactory._build_manager()._export(submission_id, folder_path)
-    _VersionManagerFactory._build_manager()._export(scenario.version, folder_path)
+            raise ExportPathAlreadyExists(str(output_zip_path), scenario_id)
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        for data_node_id in entity_ids.data_node_ids:
+            _DataManagerFactory._build_manager()._export(data_node_id, tmp_dir, include_data=include_data)
+        for task_id in entity_ids.task_ids:
+            _TaskManagerFactory._build_manager()._export(task_id, tmp_dir)
+        for sequence_id in entity_ids.sequence_ids:
+            _SequenceManagerFactory._build_manager()._export(sequence_id, tmp_dir)
+        for cycle_id in entity_ids.cycle_ids:
+            _CycleManagerFactory._build_manager()._export(cycle_id, tmp_dir)
+        for scenario_id in entity_ids.scenario_ids:
+            _ScenarioManagerFactory._build_manager()._export(scenario_id, tmp_dir)
+        for job_id in entity_ids.job_ids:
+            _JobManagerFactory._build_manager()._export(job_id, tmp_dir)
+        for submission_id in entity_ids.submission_ids:
+            _SubmissionManagerFactory._build_manager()._export(submission_id, tmp_dir)
+        _VersionManagerFactory._build_manager()._export(scenario.version, tmp_dir)
+
+        shutil.make_archive(output_filename, "zip", tmp_dir)
+
+
+def import_scenario(input_path: Union[str, pathlib.Path], override: bool = False) -> Optional[Scenario]:
+    """Import from an archive zip file containing an exported scenario into the current Taipy application.
+
+    The zip file should be created by the `taipy.import()^` method, which contains all related entities
+    of the scenario.
+    All entities should belong to the same version that is compatible with the current Taipy application version.
+
+    Parameters:
+        input_path (Union[str, pathlib.Path]): The path to the archive scenario to import.
+            If the path doesn't exist, an exception is raised.
+        override (bool): If True, override the entities if existed. Default value is False.
+
+    Return:
+        The imported scenario if the import is successful.
+
+    Raises:
+        FileNotFoundError: If the import path does not exist.
+        ImportArchiveDoesntContainAnyScenario: If the unzip folder doesn't contain any scenario.
+        ConflictedConfigurationError: If the configuration of the imported scenario is conflicted with the current one.
+    """
+    if isinstance(input_path, str):
+        zip_file_path: pathlib.Path = pathlib.Path(input_path)
+    else:
+        zip_file_path = input_path
+
+    if not zip_file_path.exists():
+        raise FileNotFoundError(f"The import archive path '{zip_file_path}' does not exist.")
+
+    entity_managers: Dict[str, Type[_Manager]] = {
+        "cycles": _CycleManagerFactory._build_manager(),
+        "cycle": _CycleManagerFactory._build_manager(),
+        "data_nodes": _DataManagerFactory._build_manager(),
+        "data_node": _DataManagerFactory._build_manager(),
+        "tasks": _TaskManagerFactory._build_manager(),
+        "task": _TaskManagerFactory._build_manager(),
+        "scenarios": _ScenarioManagerFactory._build_manager(),
+        "scenario": _ScenarioManagerFactory._build_manager(),
+        "jobs": _JobManagerFactory._build_manager(),
+        "job": _JobManagerFactory._build_manager(),
+        "submission": _SubmissionManagerFactory._build_manager(),
+        "version": _VersionManagerFactory._build_manager(),
+    }
+
+    return _ScenarioManagerFactory._build_manager()._import_scenario_and_children_entities(
+        zip_file_path, override, entity_managers
+    )
 
 
 def get_parents(
diff --git a/taipy/core/task/_task_manager_factory.py b/taipy/core/task/_task_manager_factory.py
index 0f16f009a9..61809c3e32 100644
--- a/taipy/core/task/_task_manager_factory.py
+++ b/taipy/core/task/_task_manager_factory.py
@@ -19,11 +19,10 @@
 
 
 class _TaskManagerFactory(_ManagerFactory):
-
     __REPOSITORY_MAP = {"default": _TaskFSRepository, "sql": _TaskSQLRepository}
 
     @classmethod
-    def _build_manager(cls) -> Type[_TaskManager]:  # type: ignore
+    def _build_manager(cls) -> Type[_TaskManager]:
         if cls._using_enterprise():
             task_manager = _load_fct(
                 cls._TAIPY_ENTERPRISE_CORE_MODULE + ".task._task_manager", "_TaskManager"
diff --git a/tests/core/test_taipy/test_export.py b/tests/core/test_taipy/test_export.py
index 8e47717db5..d98ee09691 100644
--- a/tests/core/test_taipy/test_export.py
+++ b/tests/core/test_taipy/test_export.py
@@ -10,21 +10,23 @@
 # specific language governing permissions and limitations under the License.
 
 import os
-import shutil
+import zipfile
 
 import pandas as pd
 import pytest
 
 import taipy.core.taipy as tp
 from taipy import Config, Frequency, Scope
-from taipy.core.exceptions import ExportFolderAlreadyExists, InvalidExportPath
+from taipy.core.exceptions import ExportPathAlreadyExists
 
 
 @pytest.fixture(scope="function", autouse=True)
-def clean_tmp_folder():
-    shutil.rmtree("./tmp", ignore_errors=True)
+def clean_export_zip_file():
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
     yield
-    shutil.rmtree("./tmp", ignore_errors=True)
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
 
 
 def plus_1(x):
@@ -57,15 +59,28 @@ def configure_test_scenario(input_data, frequency=None):
     return scenario_cfg
 
 
-def test_export_scenario_to_the_storage_folder():
+def test_export_scenario_with_and_without_zip_extension(tmp_path):
     scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+
     scenario = tp.create_scenario(scenario_cfg)
+    tp.submit(scenario)
+
+    # Export without the .zip extension should create the tmp.zip file
+    tp.export_scenario(scenario.id, f"{tmp_path}/tmp")
+    assert os.path.exists(f"{tmp_path}/tmp.zip")
+
+    os.remove(f"{tmp_path}/tmp.zip")
 
-    with pytest.raises(InvalidExportPath):
-        tp.export_scenario(scenario.id, Config.core.taipy_storage_folder)
+    # Export with the .zip extension should also create the tmp.zip file
+    tp.export_scenario(scenario.id, f"{tmp_path}/tmp.zip")
+    assert os.path.exists(f"{tmp_path}/tmp.zip")
 
+    # Export with another extension should create the tmp.<extension>.zip file
+    tp.export_scenario(scenario.id, f"{tmp_path}/tmp.tar.gz")
+    assert os.path.exists(f"{tmp_path}/tmp.tar.gz.zip")
 
-def test_export_scenario_with_cycle():
+
+def test_export_scenario_with_cycle(tmp_path):
     scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
 
     scenario = tp.create_scenario(scenario_cfg)
@@ -73,9 +88,11 @@ def test_export_scenario_with_cycle():
     jobs = submission.jobs
 
     # Export the submitted scenario
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario")
+    tp.export_scenario(scenario.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path)
 
-    assert sorted(os.listdir("./tmp/exp_scenario/data_nodes")) == sorted(
+    assert sorted(os.listdir(f"{tmp_path}/data_nodes")) == sorted(
         [
             f"{scenario.i_1.id}.json",
             f"{scenario.o_1_csv.id}.json",
@@ -84,7 +101,7 @@ def test_export_scenario_with_cycle():
             f"{scenario.o_1_json.id}.json",
         ]
     )
-    assert sorted(os.listdir("./tmp/exp_scenario/tasks")) == sorted(
+    assert sorted(os.listdir(f"{tmp_path}/tasks")) == sorted(
         [
             f"{scenario.t_1_csv.id}.json",
             f"{scenario.t_1_excel.id}.json",
@@ -92,32 +109,34 @@ def test_export_scenario_with_cycle():
             f"{scenario.t_1_json.id}.json",
         ]
     )
-    assert sorted(os.listdir("./tmp/exp_scenario/scenarios")) == sorted([f"{scenario.id}.json"])
-    assert sorted(os.listdir("./tmp/exp_scenario/jobs")) == sorted(
+    assert sorted(os.listdir(f"{tmp_path}/scenarios")) == sorted([f"{scenario.id}.json"])
+    assert sorted(os.listdir(f"{tmp_path}/jobs")) == sorted(
         [f"{jobs[0].id}.json", f"{jobs[1].id}.json", f"{jobs[2].id}.json", f"{jobs[3].id}.json"]
     )
-    assert os.listdir("./tmp/exp_scenario/submission") == [f"{submission.id}.json"]
-    assert sorted(os.listdir("./tmp/exp_scenario/cycles")) == sorted([f"{scenario.cycle.id}.json"])
+    assert os.listdir(f"{tmp_path}/submission") == [f"{submission.id}.json"]
+    assert sorted(os.listdir(f"{tmp_path}/cycles")) == sorted([f"{scenario.cycle.id}.json"])
 
 
-def test_export_scenario_without_cycle():
+def test_export_scenario_without_cycle(tmp_path):
     scenario_cfg = configure_test_scenario(1)
 
     scenario = tp.create_scenario(scenario_cfg)
     tp.submit(scenario)
 
     # Export the submitted scenario
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario")
+    tp.export_scenario(scenario.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path)
 
-    assert os.path.exists("./tmp/exp_scenario/data_nodes")
-    assert os.path.exists("./tmp/exp_scenario/tasks")
-    assert os.path.exists("./tmp/exp_scenario/scenarios")
-    assert os.path.exists("./tmp/exp_scenario/jobs")
-    assert os.path.exists("./tmp/exp_scenario/submission")
-    assert not os.path.exists("./tmp/exp_scenario/cycles")  # No cycle
+    assert os.path.exists(f"{tmp_path}/data_nodes")
+    assert os.path.exists(f"{tmp_path}/tasks")
+    assert os.path.exists(f"{tmp_path}/scenarios")
+    assert os.path.exists(f"{tmp_path}/jobs")
+    assert os.path.exists(f"{tmp_path}/submission")
+    assert not os.path.exists(f"{tmp_path}/cycles")  # No cycle
 
 
-def test_export_scenario_override_existing_files():
+def test_export_scenario_override_existing_files(tmp_path):
     scenario_1_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
     scenario_2_cfg = configure_test_scenario(2)
 
@@ -125,45 +144,54 @@ def test_export_scenario_override_existing_files():
     tp.submit(scenario_1)
 
     # Export the submitted scenario_1
-    tp.export_scenario(scenario_1.id, "./tmp/exp_scenario")
-    assert os.path.exists("./tmp/exp_scenario/data_nodes")
-    assert os.path.exists("./tmp/exp_scenario/tasks")
-    assert os.path.exists("./tmp/exp_scenario/scenarios")
-    assert os.path.exists("./tmp/exp_scenario/jobs")
-    assert os.path.exists("./tmp/exp_scenario/submission")
-    assert os.path.exists("./tmp/exp_scenario/cycles")
+    tp.export_scenario(scenario_1.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_1")
+    assert os.path.exists(f"{tmp_path}/scenario_1/data_nodes")
+    assert os.path.exists(f"{tmp_path}/scenario_1/tasks")
+    assert os.path.exists(f"{tmp_path}/scenario_1/scenarios")
+    assert os.path.exists(f"{tmp_path}/scenario_1/jobs")
+    assert os.path.exists(f"{tmp_path}/scenario_1/submission")
+    assert os.path.exists(f"{tmp_path}/scenario_1/cycles")
 
     scenario_2 = tp.create_scenario(scenario_2_cfg)
     tp.submit(scenario_2)
 
-    # Export the submitted scenario_2 to the same folder should raise an error
-    with pytest.raises(ExportFolderAlreadyExists):
-        tp.export_scenario(scenario_2.id, "./tmp/exp_scenario")
+    # Export the submitted scenario_2 to the same path should raise an error
+    with pytest.raises(ExportPathAlreadyExists):
+        tp.export_scenario(scenario_2.id, "tmp.zip")
 
     # Export the submitted scenario_2 without a cycle and override the existing files
-    tp.export_scenario(scenario_2.id, "./tmp/exp_scenario", override=True)
-    assert os.path.exists("./tmp/exp_scenario/data_nodes")
-    assert os.path.exists("./tmp/exp_scenario/tasks")
-    assert os.path.exists("./tmp/exp_scenario/scenarios")
-    assert os.path.exists("./tmp/exp_scenario/jobs")
-    assert os.path.exists("./tmp/exp_scenario/submission")
-    # The cycles folder should be removed when overriding
-    assert not os.path.exists("./tmp/exp_scenario/cycles")
-
-
-def test_export_scenario_filesystem_with_data():
+    tp.export_scenario(scenario_2.id, "tmp.zip", override=True)
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_2")
+    assert os.path.exists(f"{tmp_path}/scenario_2/data_nodes")
+    assert os.path.exists(f"{tmp_path}/scenario_2/tasks")
+    assert os.path.exists(f"{tmp_path}/scenario_2/scenarios")
+    assert os.path.exists(f"{tmp_path}/scenario_2/jobs")
+    assert os.path.exists(f"{tmp_path}/scenario_2/submission")
+    # The cycles folder should not exists since the new scenario does not have a cycle
+    assert not os.path.exists(f"{tmp_path}/scenario_2/cycles")
+
+
+def test_export_scenario_filesystem_with_data(tmp_path):
     scenario_cfg = configure_test_scenario(1)
     scenario = tp.create_scenario(scenario_cfg)
     tp.submit(scenario)
 
     # Export scenario without data
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario")
-    assert not os.path.exists("./tmp/exp_scenario/user_data")
+    tp.export_scenario(scenario.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_without_data")
+    assert not os.path.exists(f"{tmp_path}/scenario_without_data/user_data")
 
     # Export scenario with data
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario", include_data=True, override=True)
-    assert os.path.exists("./tmp/exp_scenario/user_data")
-    data_files = [f for _, _, files in os.walk("./tmp/exp_scenario/user_data") for f in files]
+    tp.export_scenario(scenario.id, "tmp.zip", include_data=True, override=True)
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_with_data")
+    assert os.path.exists(f"{tmp_path}/scenario_with_data/user_data")
+
+    data_files = [f for _, _, files in os.walk(f"{tmp_path}/scenario_with_data/user_data") for f in files]
     assert sorted(data_files) == sorted(
         [
             f"{scenario.i_1.id}.p",
@@ -188,6 +216,6 @@ def test_export_non_file_based_data_node_raise_warning(caplog):
     tp.submit(scenario)
 
     # Export scenario with in-memory data node
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario", include_data=True)
+    tp.export_scenario(scenario.id, "tmp.zip", include_data=True)
     expected_warning = f"Data node {scenario.o_mem.id} is not a file-based data node and the data will not be exported"
     assert expected_warning in caplog.text
diff --git a/tests/core/test_taipy/test_export_with_sql_repo.py b/tests/core/test_taipy/test_export_with_sql_repo.py
index b5a4700abe..7ca34c7d61 100644
--- a/tests/core/test_taipy/test_export_with_sql_repo.py
+++ b/tests/core/test_taipy/test_export_with_sql_repo.py
@@ -10,21 +10,23 @@
 # specific language governing permissions and limitations under the License.
 
 import os
-import shutil
+import zipfile
 
 import pandas as pd
 import pytest
 
 import taipy.core.taipy as tp
 from taipy import Config, Frequency, Scope
-from taipy.core.exceptions import ExportFolderAlreadyExists, InvalidExportPath
+from taipy.core.exceptions import ExportPathAlreadyExists
 
 
 @pytest.fixture(scope="function", autouse=True)
-def clean_tmp_folder():
-    shutil.rmtree("./tmp", ignore_errors=True)
+def clean_export_zip_file():
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
     yield
-    shutil.rmtree("./tmp", ignore_errors=True)
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
 
 
 def plus_1(x):
@@ -57,15 +59,7 @@ def configure_test_scenario(input_data, frequency=None):
     return scenario_cfg
 
 
-def test_export_scenario_to_the_storage_folder(init_sql_repo):
-    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
-    scenario = tp.create_scenario(scenario_cfg)
-
-    with pytest.raises(InvalidExportPath):
-        tp.export_scenario(scenario.id, Config.core.taipy_storage_folder)
-
-
-def test_export_scenario_with_cycle(init_sql_repo):
+def test_export_scenario_with_cycle(tmp_path, init_sql_repo):
     scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
 
     scenario = tp.create_scenario(scenario_cfg)
@@ -73,9 +67,11 @@ def test_export_scenario_with_cycle(init_sql_repo):
     jobs = submission.jobs
 
     # Export the submitted scenario
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario")
+    tp.export_scenario(scenario.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path)
 
-    assert sorted(os.listdir("./tmp/exp_scenario/data_node")) == sorted(
+    assert sorted(os.listdir(f"{tmp_path}/data_node")) == sorted(
         [
             f"{scenario.i_1.id}.json",
             f"{scenario.o_1_csv.id}.json",
@@ -84,7 +80,7 @@ def test_export_scenario_with_cycle(init_sql_repo):
             f"{scenario.o_1_json.id}.json",
         ]
     )
-    assert sorted(os.listdir("./tmp/exp_scenario/task")) == sorted(
+    assert sorted(os.listdir(f"{tmp_path}/task")) == sorted(
         [
             f"{scenario.t_1_csv.id}.json",
             f"{scenario.t_1_excel.id}.json",
@@ -92,32 +88,34 @@ def test_export_scenario_with_cycle(init_sql_repo):
             f"{scenario.t_1_json.id}.json",
         ]
     )
-    assert sorted(os.listdir("./tmp/exp_scenario/scenario")) == sorted([f"{scenario.id}.json"])
-    assert sorted(os.listdir("./tmp/exp_scenario/job")) == sorted(
+    assert sorted(os.listdir(f"{tmp_path}/scenario")) == sorted([f"{scenario.id}.json"])
+    assert sorted(os.listdir(f"{tmp_path}/job")) == sorted(
         [f"{jobs[0].id}.json", f"{jobs[1].id}.json", f"{jobs[2].id}.json", f"{jobs[3].id}.json"]
     )
-    assert os.listdir("./tmp/exp_scenario/submission") == [f"{submission.id}.json"]
-    assert sorted(os.listdir("./tmp/exp_scenario/cycle")) == sorted([f"{scenario.cycle.id}.json"])
+    assert os.listdir(f"{tmp_path}/submission") == [f"{submission.id}.json"]
+    assert sorted(os.listdir(f"{tmp_path}/cycle")) == sorted([f"{scenario.cycle.id}.json"])
 
 
-def test_export_scenario_without_cycle(init_sql_repo):
+def test_export_scenario_without_cycle(tmp_path, init_sql_repo):
     scenario_cfg = configure_test_scenario(1)
 
     scenario = tp.create_scenario(scenario_cfg)
     tp.submit(scenario)
 
     # Export the submitted scenario
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario")
+    tp.export_scenario(scenario.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path)
 
-    assert os.path.exists("./tmp/exp_scenario/data_node")
-    assert os.path.exists("./tmp/exp_scenario/task")
-    assert os.path.exists("./tmp/exp_scenario/scenario")
-    assert os.path.exists("./tmp/exp_scenario/job")
-    assert os.path.exists("./tmp/exp_scenario/submission")
-    assert not os.path.exists("./tmp/exp_scenario/cycle")  # No cycle
+    assert os.path.exists(f"{tmp_path}/data_node")
+    assert os.path.exists(f"{tmp_path}/task")
+    assert os.path.exists(f"{tmp_path}/scenario")
+    assert os.path.exists(f"{tmp_path}/job")
+    assert os.path.exists(f"{tmp_path}/submission")
+    assert not os.path.exists(f"{tmp_path}/cycle")  # No cycle
 
 
-def test_export_scenario_override_existing_files(init_sql_repo):
+def test_export_scenario_override_existing_files(tmp_path, init_sql_repo):
     scenario_1_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
     scenario_2_cfg = configure_test_scenario(2)
 
@@ -125,45 +123,54 @@ def test_export_scenario_override_existing_files(init_sql_repo):
     tp.submit(scenario_1)
 
     # Export the submitted scenario_1
-    tp.export_scenario(scenario_1.id, "./tmp/exp_scenario")
-    assert os.path.exists("./tmp/exp_scenario/data_node")
-    assert os.path.exists("./tmp/exp_scenario/task")
-    assert os.path.exists("./tmp/exp_scenario/scenario")
-    assert os.path.exists("./tmp/exp_scenario/job")
-    assert os.path.exists("./tmp/exp_scenario/submission")
-    assert os.path.exists("./tmp/exp_scenario/cycle")
+    tp.export_scenario(scenario_1.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_1")
+    assert os.path.exists(f"{tmp_path}/scenario_1/data_node")
+    assert os.path.exists(f"{tmp_path}/scenario_1/task")
+    assert os.path.exists(f"{tmp_path}/scenario_1/scenario")
+    assert os.path.exists(f"{tmp_path}/scenario_1/job")
+    assert os.path.exists(f"{tmp_path}/scenario_1/submission")
+    assert os.path.exists(f"{tmp_path}/scenario_1/cycle")
 
     scenario_2 = tp.create_scenario(scenario_2_cfg)
     tp.submit(scenario_2)
 
     # Export the submitted scenario_2 to the same folder should raise an error
-    with pytest.raises(ExportFolderAlreadyExists):
-        tp.export_scenario(scenario_2.id, "./tmp/exp_scenario")
+    with pytest.raises(ExportPathAlreadyExists):
+        tp.export_scenario(scenario_2.id, "tmp.zip")
 
     # Export the submitted scenario_2 without a cycle and override the existing files
-    tp.export_scenario(scenario_2.id, "./tmp/exp_scenario", override=True)
-    assert os.path.exists("./tmp/exp_scenario/data_node")
-    assert os.path.exists("./tmp/exp_scenario/task")
-    assert os.path.exists("./tmp/exp_scenario/scenario")
-    assert os.path.exists("./tmp/exp_scenario/job")
-    assert os.path.exists("./tmp/exp_scenario/submission")
-    # The cycles folder should be removed when overriding
-    assert not os.path.exists("./tmp/exp_scenario/cycle")
-
-
-def test_export_scenario_filesystem_with_data(init_sql_repo):
+    tp.export_scenario(scenario_2.id, "tmp.zip", override=True)
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_2")
+    assert os.path.exists(f"{tmp_path}/scenario_2/data_node")
+    assert os.path.exists(f"{tmp_path}/scenario_2/task")
+    assert os.path.exists(f"{tmp_path}/scenario_2/scenario")
+    assert os.path.exists(f"{tmp_path}/scenario_2/job")
+    assert os.path.exists(f"{tmp_path}/scenario_2/submission")
+    # The cycles folder should not exists since the new scenario does not have a cycle
+    assert not os.path.exists(f"{tmp_path}/scenario_2/cycle")
+
+
+def test_export_scenario_sql_repo_with_data(tmp_path, init_sql_repo):
     scenario_cfg = configure_test_scenario(1)
     scenario = tp.create_scenario(scenario_cfg)
     tp.submit(scenario)
 
     # Export scenario without data
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario")
-    assert not os.path.exists("./tmp/exp_scenario/user_data")
+    tp.export_scenario(scenario.id, "tmp.zip")
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_without_data")
+    assert not os.path.exists(f"{tmp_path}/scenario_without_data/user_data")
 
     # Export scenario with data
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario", include_data=True, override=True)
-    assert os.path.exists("./tmp/exp_scenario/user_data")
-    data_files = [f for _, _, files in os.walk("./tmp/exp_scenario/user_data") for f in files]
+    tp.export_scenario(scenario.id, "tmp.zip", include_data=True, override=True)
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path / "scenario_with_data")
+    assert os.path.exists(f"{tmp_path}/scenario_with_data/user_data")
+
+    data_files = [f for _, _, files in os.walk(f"{tmp_path}/scenario_with_data/user_data") for f in files]
     assert sorted(data_files) == sorted(
         [
             f"{scenario.i_1.id}.p",
@@ -188,6 +195,6 @@ def test_export_non_file_based_data_node_raise_warning(init_sql_repo, caplog):
     tp.submit(scenario)
 
     # Export scenario with in-memory data node
-    tp.export_scenario(scenario.id, "./tmp/exp_scenario", include_data=True)
+    tp.export_scenario(scenario.id, "tmp.zip", include_data=True)
     expected_warning = f"Data node {scenario.o_mem.id} is not a file-based data node and the data will not be exported"
     assert expected_warning in caplog.text
diff --git a/tests/core/test_taipy/test_import.py b/tests/core/test_taipy/test_import.py
new file mode 100644
index 0000000000..742680dcab
--- /dev/null
+++ b/tests/core/test_taipy/test_import.py
@@ -0,0 +1,212 @@
+# Copyright 2021-2024 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+
+import os
+import shutil
+import zipfile
+
+import pandas as pd
+import pytest
+
+import taipy.core.taipy as tp
+from taipy import Config, Frequency, Scope
+from taipy.core._version._version_manager import _VersionManager
+from taipy.core.cycle._cycle_manager import _CycleManager
+from taipy.core.data._data_manager import _DataManager
+from taipy.core.exceptions.exceptions import (
+    ConflictedConfigurationError,
+    ImportArchiveDoesntContainAnyScenario,
+    ImportScenarioDoesntHaveAVersion,
+)
+from taipy.core.job._job_manager import _JobManager
+from taipy.core.scenario._scenario_manager import _ScenarioManager
+from taipy.core.submission._submission_manager import _SubmissionManager
+from taipy.core.task._task_manager import _TaskManager
+
+
+@pytest.fixture(scope="function", autouse=True)
+def clean_export_zip_file():
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
+    yield
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
+
+
+def plus_1(x):
+    return x + 1
+
+
+def plus_1_dataframe(x):
+    return pd.DataFrame({"output": [x + 1]})
+
+
+def configure_test_scenario(input_data, frequency=None):
+    input_cfg = Config.configure_data_node(
+        id=f"i_{input_data}", storage_type="pickle", scope=Scope.SCENARIO, default_data=input_data
+    )
+    csv_output_cfg = Config.configure_data_node(id=f"o_{input_data}_csv", storage_type="csv")
+    excel_output_cfg = Config.configure_data_node(id=f"o_{input_data}_excel", storage_type="excel")
+    parquet_output_cfg = Config.configure_data_node(id=f"o_{input_data}_parquet", storage_type="parquet")
+    json_output_cfg = Config.configure_data_node(id=f"o_{input_data}_json", storage_type="json")
+
+    csv_task_cfg = Config.configure_task(f"t_{input_data}_csv", plus_1_dataframe, input_cfg, csv_output_cfg)
+    excel_task_cfg = Config.configure_task(f"t_{input_data}_excel", plus_1_dataframe, input_cfg, excel_output_cfg)
+    parquet_task_cfg = Config.configure_task(f"t_{input_data}_parquet", plus_1_dataframe, input_cfg, parquet_output_cfg)
+    json_task_cfg = Config.configure_task(f"t_{input_data}_json", plus_1, input_cfg, json_output_cfg)
+    scenario_cfg = Config.configure_scenario(
+        id=f"s_{input_data}",
+        task_configs=[csv_task_cfg, excel_task_cfg, parquet_task_cfg, json_task_cfg],
+        frequency=frequency,
+    )
+
+    return scenario_cfg
+
+
+def export_test_scenario(scenario_cfg, export_path="tmp.zip", override=False, include_data=False):
+    scenario = tp.create_scenario(scenario_cfg)
+    tp.submit(scenario)
+
+    # Export the submitted scenario
+    tp.export_scenario(scenario.id, export_path, override, include_data)
+    return scenario
+
+
+def test_import_scenario_without_data(init_managers):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    scenario = export_test_scenario(scenario_cfg)
+
+    init_managers()
+
+    assert _ScenarioManager._get_all() == []
+    imported_scenario = tp.import_scenario("tmp.zip")
+
+    # The imported scenario should be the same as the exported scenario
+    assert _ScenarioManager._get_all() == [imported_scenario]
+    assert imported_scenario == scenario
+
+    # All entities belonging to the scenario should be imported
+    assert len(_CycleManager._get_all()) == 1
+    assert len(_TaskManager._get_all()) == 4
+    assert len(_DataManager._get_all()) == 5
+    assert len(_JobManager._get_all()) == 4
+    assert len(_SubmissionManager._get_all()) == 1
+    assert len(_VersionManager._get_all()) == 1
+
+
+def test_import_scenario_with_data(init_managers):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg, include_data=True)
+
+    init_managers()
+
+    assert _ScenarioManager._get_all() == []
+    imported_scenario = tp.import_scenario("tmp.zip")
+
+    # All data of all data nodes should be imported
+    assert all(os.path.exists(dn.path) for dn in imported_scenario.data_nodes.values())
+
+
+def test_import_scenario_when_entities_are_already_existed_should_rollback(caplog):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg)
+
+    caplog.clear()
+
+    _CycleManager._delete_all()
+    _TaskManager._delete_all()
+    _DataManager._delete_all()
+    _JobManager._delete_all()
+    _ScenarioManager._delete_all()
+
+    assert len(_CycleManager._get_all()) == 0
+    assert len(_TaskManager._get_all()) == 0
+    assert len(_DataManager._get_all()) == 0
+    assert len(_JobManager._get_all()) == 0
+    assert len(_SubmissionManager._get_all()) == 1  # Keep the submission entity to test the rollback
+    submission_id = _SubmissionManager._get_all()[0].id
+    assert len(_ScenarioManager._get_all()) == 0
+
+    # Import the scenario when the old entities still exist
+    imported_entity = tp.import_scenario("tmp.zip")
+    assert imported_entity is None
+    assert all(log.levelname in ["ERROR", "INFO"] for log in caplog.records)
+    assert "An error occurred during the import" in caplog.text
+    assert f"{submission_id} already exists. Please use the 'override' parameter to override it" in caplog.text
+
+    # No entity should be imported and the old entities should be kept
+    assert len(_CycleManager._get_all()) == 0
+    assert len(_TaskManager._get_all()) == 0
+    assert len(_DataManager._get_all()) == 0
+    assert len(_JobManager._get_all()) == 0
+    assert len(_SubmissionManager._get_all()) == 1  # Keep the submission entity to test the rollback
+    assert len(_ScenarioManager._get_all()) == 0
+
+    caplog.clear()
+
+    # Import with override flag
+    tp.import_scenario("tmp.zip", override=True)
+    assert all(log.levelname in ["WARNING", "INFO"] for log in caplog.records)
+    assert f"{submission_id} already exists and will be overridden" in caplog.text
+
+    # The scenario is imported and overridden the old one
+    assert len(_ScenarioManager._get_all()) == 1
+    assert len(_CycleManager._get_all()) == 1
+    assert len(_TaskManager._get_all()) == 4
+    assert len(_DataManager._get_all()) == 5
+    assert len(_JobManager._get_all()) == 4
+    assert len(_SubmissionManager._get_all()) == 1
+    assert len(_VersionManager._get_all()) == 1
+
+
+def test_import_incompatible_scenario(init_managers):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg)
+
+    Config.unblock_update()
+
+    # Configure a new dn to make the exported version incompatible
+    Config.configure_data_node("new_dn")
+
+    with pytest.raises(ConflictedConfigurationError):
+        tp.import_scenario("tmp.zip")
+
+
+def test_import_a_non_exists_folder():
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg)
+
+    with pytest.raises(FileNotFoundError):
+        tp.import_scenario("non_exists_folder")
+
+
+def test_import_an_empty_archive(tmpdir_factory):
+    empty_folder = tmpdir_factory.mktemp("empty_folder").strpath
+    shutil.make_archive("tmp", "zip", empty_folder)
+
+    with pytest.raises(ImportArchiveDoesntContainAnyScenario):
+        tp.import_scenario("tmp.zip")
+
+
+def test_import_with_no_version(tmp_path):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg)
+
+    # Extract the zip,
+    with zipfile.ZipFile("./tmp.zip") as zip_file:
+        zip_file.extractall(tmp_path)
+    # remove the version,
+    shutil.rmtree(f"{tmp_path}/version")
+    # and archive the scenario without the version again
+    shutil.make_archive("tmp", "zip", tmp_path)
+
+    with pytest.raises(ImportScenarioDoesntHaveAVersion):
+        tp.import_scenario("tmp.zip")
diff --git a/tests/core/test_taipy/test_import_with_sql_repo.py b/tests/core/test_taipy/test_import_with_sql_repo.py
new file mode 100644
index 0000000000..c1ef8a73be
--- /dev/null
+++ b/tests/core/test_taipy/test_import_with_sql_repo.py
@@ -0,0 +1,174 @@
+# Copyright 2021-2024 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+
+import os
+
+import pandas as pd
+import pytest
+
+import taipy.core.taipy as tp
+from taipy import Config, Frequency, Scope
+from taipy.core._version._version_manager import _VersionManager
+from taipy.core.cycle._cycle_manager import _CycleManager
+from taipy.core.data._data_manager import _DataManager
+from taipy.core.exceptions.exceptions import ConflictedConfigurationError
+from taipy.core.job._job_manager import _JobManager
+from taipy.core.scenario._scenario_manager import _ScenarioManager
+from taipy.core.submission._submission_manager import _SubmissionManager
+from taipy.core.task._task_manager import _TaskManager
+
+
+@pytest.fixture(scope="function", autouse=True)
+def clean_export_zip_file():
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
+    yield
+    if os.path.exists("./tmp.zip"):
+        os.remove("./tmp.zip")
+
+
+def plus_1(x):
+    return x + 1
+
+
+def plus_1_dataframe(x):
+    return pd.DataFrame({"output": [x + 1]})
+
+
+def configure_test_scenario(input_data, frequency=None):
+    input_cfg = Config.configure_data_node(
+        id=f"i_{input_data}", storage_type="pickle", scope=Scope.SCENARIO, default_data=input_data
+    )
+    csv_output_cfg = Config.configure_data_node(id=f"o_{input_data}_csv", storage_type="csv")
+    excel_output_cfg = Config.configure_data_node(id=f"o_{input_data}_excel", storage_type="excel")
+    parquet_output_cfg = Config.configure_data_node(id=f"o_{input_data}_parquet", storage_type="parquet")
+    json_output_cfg = Config.configure_data_node(id=f"o_{input_data}_json", storage_type="json")
+
+    csv_task_cfg = Config.configure_task(f"t_{input_data}_csv", plus_1_dataframe, input_cfg, csv_output_cfg)
+    excel_task_cfg = Config.configure_task(f"t_{input_data}_excel", plus_1_dataframe, input_cfg, excel_output_cfg)
+    parquet_task_cfg = Config.configure_task(f"t_{input_data}_parquet", plus_1_dataframe, input_cfg, parquet_output_cfg)
+    json_task_cfg = Config.configure_task(f"t_{input_data}_json", plus_1, input_cfg, json_output_cfg)
+    scenario_cfg = Config.configure_scenario(
+        id=f"s_{input_data}",
+        task_configs=[csv_task_cfg, excel_task_cfg, parquet_task_cfg, json_task_cfg],
+        frequency=frequency,
+    )
+
+    return scenario_cfg
+
+
+def export_test_scenario(scenario_cfg, export_path="tmp.zip", override=False, include_data=False):
+    scenario = tp.create_scenario(scenario_cfg)
+    tp.submit(scenario)
+
+    # Export the submitted scenario
+    tp.export_scenario(scenario.id, export_path, override, include_data)
+    return scenario
+
+
+def test_import_scenario_without_data(init_sql_repo, init_managers):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    scenario = export_test_scenario(scenario_cfg)
+
+    init_managers()
+
+    assert _ScenarioManager._get_all() == []
+    imported_scenario = tp.import_scenario("tmp.zip")
+
+    # The imported scenario should be the same as the exported scenario
+    assert _ScenarioManager._get_all() == [imported_scenario]
+    assert imported_scenario == scenario
+
+    # All entities belonging to the scenario should be imported
+    assert len(_CycleManager._get_all()) == 1
+    assert len(_TaskManager._get_all()) == 4
+    assert len(_DataManager._get_all()) == 5
+    assert len(_JobManager._get_all()) == 4
+    assert len(_SubmissionManager._get_all()) == 1
+    assert len(_VersionManager._get_all()) == 1
+
+
+def test_import_scenario_with_data(init_sql_repo, init_managers):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg, include_data=True)
+
+    init_managers()
+
+    assert _ScenarioManager._get_all() == []
+    imported_scenario = tp.import_scenario("tmp.zip")
+
+    # All data of all data nodes should be imported
+    assert all(os.path.exists(dn.path) for dn in imported_scenario.data_nodes.values())
+
+
+def test_import_scenario_when_entities_are_already_existed_should_rollback(init_sql_repo, caplog):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg)
+
+    caplog.clear()
+
+    _CycleManager._delete_all()
+    _TaskManager._delete_all()
+    _DataManager._delete_all()
+    _JobManager._delete_all()
+    _ScenarioManager._delete_all()
+
+    assert len(_CycleManager._get_all()) == 0
+    assert len(_TaskManager._get_all()) == 0
+    assert len(_DataManager._get_all()) == 0
+    assert len(_JobManager._get_all()) == 0
+    assert len(_SubmissionManager._get_all()) == 1  # Keep the submission entity to test the rollback
+    submission_id = _SubmissionManager._get_all()[0].id
+    assert len(_ScenarioManager._get_all()) == 0
+
+    # Import the scenario when the old entities still exist
+    imported_entity = tp.import_scenario("tmp.zip")
+    assert imported_entity is None
+    assert all(log.levelname in ["ERROR", "INFO"] for log in caplog.records)
+    assert "An error occurred during the import" in caplog.text
+    assert f"{submission_id} already exists. Please use the 'override' parameter to override it" in caplog.text
+
+    # No entity should be imported and the old entities should be kept
+    assert len(_CycleManager._get_all()) == 0
+    assert len(_TaskManager._get_all()) == 0
+    assert len(_DataManager._get_all()) == 0
+    assert len(_JobManager._get_all()) == 0
+    assert len(_SubmissionManager._get_all()) == 1  # Keep the submission entity to test the rollback
+    assert len(_ScenarioManager._get_all()) == 0
+
+    caplog.clear()
+
+    # Import with override flag
+    tp.import_scenario("tmp.zip", override=True)
+    assert all(log.levelname in ["WARNING", "INFO"] for log in caplog.records)
+    assert f"{submission_id} already exists and will be overridden" in caplog.text
+
+    # The scenario is imported and overridden the old one
+    assert len(_ScenarioManager._get_all()) == 1
+    assert len(_CycleManager._get_all()) == 1
+    assert len(_TaskManager._get_all()) == 4
+    assert len(_DataManager._get_all()) == 5
+    assert len(_JobManager._get_all()) == 4
+    assert len(_SubmissionManager._get_all()) == 1
+    assert len(_VersionManager._get_all()) == 1
+
+
+def test_import_incompatible_scenario(init_sql_repo, init_managers):
+    scenario_cfg = configure_test_scenario(1, frequency=Frequency.DAILY)
+    export_test_scenario(scenario_cfg)
+
+    Config.unblock_update()
+
+    # Configure a new dn to make the exported version incompatible
+    Config.configure_data_node("new_dn")
+
+    with pytest.raises(ConflictedConfigurationError):
+        tp.import_scenario("tmp.zip")


Code review comments:
diff_hunk: @@ -450,3 +457,85 @@ def _get_by_config_id(cls, config_id: str, version_number: Optional[str] = None)
         for fil in filters:
             fil.update({"config_id": config_id})
         return cls._repository._load_all(filters)
+
+    @classmethod
+    def _import_scenario_and_children_entities(
+        cls,
+        zip_file_path: pathlib.Path,
+        override: bool,
+        entity_managers: Dict[str, Type[_Manager]],
+    ) -> Optional[Scenario]:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            with zipfile.ZipFile(zip_file_path) as zip_file:
+                zip_file.extractall(tmp_dir)
+
+            tmp_dir_path = pathlib.Path(tmp_dir)
+
+            if not ((tmp_dir_path / "scenarios").exists() or (tmp_dir_path / "scenario").exists()):
+                raise ImportArchiveDoesntContainAnyScenario(zip_file_path)
+
+            if not (tmp_dir_path / "version").exists():
+                raise ImportScenarioDoesntHaveAVersion(zip_file_path)
+
+            # Import the version to check for compatibility
+            entity_managers["version"]._import(next((tmp_dir_path / "version").iterdir()), "")
+
+            valid_entity_folders = list(entity_managers.keys())
+            valid_data_folder = Config.core.storage_folder
+
+            imported_scenario = None
+            imported_entities: Dict[str, List] = {}
+
+            for entity_folder in tmp_dir_path.iterdir():
+                if not entity_folder.is_dir() or entity_folder.name not in valid_entity_folders + [valid_data_folder]:
+                    cls._logger.warning(f"{entity_folder} is not a valid Taipy folder and will not be imported.")
+                    continue
+
+            try:
+                for entity_type in valid_entity_folders:
+                    # Skip the version folder as it is already handled
+                    if entity_type == "version":
+                        continue
+
+                    entity_folder = tmp_dir_path / entity_type
+                    if not entity_folder.exists():
+                        continue
+
+                    manager = entity_managers[entity_type]
+                    imported_entities[entity_type] = []
+
+                    for entity_file in entity_folder.iterdir():
+                        # Check if the to-be-imported entity already exists
+                        entity_id = entity_file.stem
+                        if manager._exists(entity_id):
+                            if override:
+                                cls._logger.warning(f"{entity_id} already exists and will be overridden.")
+                            else:
+                                cls._logger.error(
+                                    f"{entity_id} already exists. Please use the 'override' parameter to override it."
+                                )
+                                raise EntitiesToBeImportAlredyExist(zip_file_path)
+
+                        # Import the entity
+                        imported_entity = manager._import(
+                            entity_file,
+                            version=_VersionManagerFactory._build_manager()._get_latest_version(),
+                            data_folder=tmp_dir_path / valid_data_folder,
+                        )
+
+                        imported_entities[entity_type].append(imported_entity.id)
+                        if entity_type in ["scenario", "scenarios"]:
+                            imported_scenario = imported_entity
+            except Exception as err:
+                cls._logger.error(f"An error occurred during the import: {err}. Rollback the import.")
+
+                # Rollback the import
+                for entity_type, entity_ids in list(imported_entities.items())[::-1]:
+                    manager = entity_managers[entity_type]
+                    for entity_id in entity_ids:
+                        if manager._exists(entity_id):
+                            manager._delete(entity_id)
+                return None
path: taipy/core/scenario/_scenario_manager.py
text: @author:
Here, when there is any Exception, we will return None. Do we want to raise the Exception again here, after the rollback?
```suggestion
                raise err
```

@user1:
Yes, I would raise the error again.