Feature/poc cache study dao

antarest/study/dao/cache_study_dao.py

@@ -0,0 +1,43 @@
+from typing import Any, Dict, Optional
+
+from antarest.core.interfaces.cache import ICache
+from antarest.study.dao.study_dao import StudiesDAO, StudyDAO
+
+
+class CacheStudyDAO(StudyDAO):
+    """
+    Implementation which preferably reads from cache,
+    but delegates to a possibly slower implementation if
+    data is not in cache.
+
+    All cache management details are implemented here.
+    """
+
+    def __init__(self, study_id: str, cache: ICache, delegate_factory: StudiesDAO):
+        self._study_id = study_id
+        self._cache = cache
+        self._delegate_factory = delegate_factory
+        self._delegate: Optional[StudyDAO] = None
+
+    def _get_delegate(self) -> StudyDAO:
+        if self._delegate is None:
+            self._delegate = self._delegate_factory.get_study(self._study_id)
+        return self._delegate
+
+    def get_all_areas_ui_info(self) -> Dict[str, Any]:
+        cache_id = f"{self._study_id}-areas-ui"
+        from_cache = self._cache.get(cache_id)
+        if from_cache is not None:
+            return from_cache
+        ui_info = self._get_delegate().get_all_areas_ui_info()
+        self._cache.put(cache_id, ui_info)
+        return ui_info
+
+
+class CacheStudiesDAO(StudiesDAO):
+    def __init__(self, cache: ICache, delegate_factory: StudiesDAO):
+        self._cache = cache
+        self._delegate_factory = delegate_factory
+
+    def get_study(self, study_id: str) -> StudyDAO:
+        return CacheStudyDAO(study_id, self._cache, self._delegate_factory)

antarest/study/dao/file_study_dao.py

@@ -0,0 +1,81 @@
+from typing import TYPE_CHECKING, Any, Dict, Sequence
+
+from antarest.study.dao.study_dao import StudiesDAO, StudyDAO
+
+if TYPE_CHECKING:
+    from antarest.study.service import StudyService
+
+from antarest.study.storage.rawstudy.model.filesystem.config.area import UIProperties
+from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy
+
+
+def _get_ui_info_map(file_study: FileStudy, area_ids: Sequence[str]) -> Dict[str, Any]:
+    """
+    Get the UI information (a JSON object) for each selected Area.
+
+    Args:
+        file_study: A file study from which the configuration can be read.
+        area_ids: List of selected area IDs.
+
+    Returns:
+        Dictionary where keys are IDs, and values are UI objects.
+
+    Raises:
+        ChildNotFoundError: if one of the Area IDs is not found in the configuration.
+    """
+    # If there is no ID, it is better to return an empty dictionary
+    # instead of raising an obscure exception.
+    if not area_ids:
+        return {}
+
+    ui_info_map = file_study.tree.get(["input", "areas", ",".join(area_ids), "ui"])
+
+    # If there is only one ID in the `area_ids`, the result returned from
+    # the `file_study.tree.get` call will be a single UI object.
+    # On the other hand, if there are multiple values in `area_ids`,
+    # the result will be a dictionary where the keys are the IDs,
+    # and the values are the corresponding UI objects.
+    if len(area_ids) == 1:
+        ui_info_map = {area_ids[0]: ui_info_map}
+
+    # Convert to UIProperties to ensure that the UI object is valid.
+    ui_info_map = {area_id: UIProperties(**ui_info).to_config() for area_id, ui_info in ui_info_map.items()}
+
+    return ui_info_map
+
+
+class FileStudyDAO(StudyDAO):
+    """
+    Implementation which fetches and writes study data from file tree representation.
+    """
+
+    def __init__(self, file_study: FileStudy):
+        self._file_study = file_study
+
+    def get_all_areas_ui_info(self) -> Dict[str, Any]:
+        """
+        Retrieve information about all areas' user interface (UI) from the study.
+
+        Args:
+            study: The raw study object containing the study's data.
+
+        Returns:
+            A dictionary containing information about the user interface for the areas.
+
+        Raises:
+            ChildNotFoundError: if one of the Area IDs is not found in the configuration.
+        """
+        study = self._file_study
+        area_ids = list(study.config.areas)
+        res = _get_ui_info_map(study, area_ids)
+        return res
+
+
+class FileStudiesDAO(StudiesDAO):
+    def __init__(self, study_service: "StudyService"):
+        self._study_service = study_service
+
+    def get_study(self, study_id: str) -> StudyDAO:
+        study_metadata = self._study_service.get_study(study_id)
+        file_study = self._study_service.storage_service.get_storage(study_metadata).get_raw(study_metadata)
+        return FileStudyDAO(file_study)

antarest/study/dao/study_dao.py

@@ -0,0 +1,18 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+
+class StudyDAO(ABC):
+    """
+    Interface to access and mofify study data.
+    """
+
+    @abstractmethod
+    def get_all_areas_ui_info(self) -> Dict[str, Any]:
+        raise NotImplementedError()
+
+
+class StudiesDAO(ABC):
+    @abstractmethod
+    def get_study(self, study_id: str) -> StudyDAO:
+        raise NotImplementedError()

antarest/study/service.py

@@ -96,6 +96,9 @@
     XpansionCandidateDTO,
     XpansionManager,
 )
+from antarest.study.dao.cache_study_dao import CacheStudiesDAO
+from antarest.study.dao.file_study_dao import FileStudiesDAO
+from antarest.study.dao.study_dao import StudiesDAO
 from antarest.study.model import (
     DEFAULT_WORKSPACE_NAME,
     NEW_DEFAULT_STUDY_VERSION,
@@ -355,6 +358,12 @@ def __init__(
         self.config = config
         self.on_deletion_callbacks: t.List[t.Callable[[str], None]] = []
 
+        self.studies_dao = self._create_studies_dao()
+
+    def _create_studies_dao(self) -> StudiesDAO:
+        delegate = FileStudiesDAO(self)
+        return CacheStudiesDAO(self.cache_service, delegate)
+
     def add_on_deletion_callback(self, callback: t.Callable[[str], None]) -> None:
         self.on_deletion_callbacks.append(callback)
 
@@ -1806,7 +1815,9 @@ def get_all_areas(
     ) -> t.Union[t.List[AreaInfoDTO], t.Dict[str, t.Any]]:
         study = self.get_study(uuid)
         assert_permission(params.user, study, StudyPermissionType.READ)
-        return self.areas.get_all_areas_ui_info(study) if ui else self.areas.get_all_areas(study, area_type)
+        if ui:
+            return self.studies_dao.get_study(uuid).get_all_areas_ui_info()
+        return self.areas.get_all_areas(study, area_type)
 
     def get_all_links(
         self,