refactor(api): refactor and rename the exceptions related to .ini file configuration

antarest/core/exceptions.py

@@ -1,3 +1,4 @@
+import re
 from http import HTTPStatus
 from typing import Optional
 
@@ -8,64 +9,227 @@ class ShouldNotHappenException(Exception):
     pass
 
 
-class STStorageFieldsNotFoundError(HTTPException):
-    """Fields of the short-term storage are not found"""
+# ============================================================
+# Exceptions related to the study configuration (`.ini` files)
+# ============================================================
 
-    def __init__(self, storage_id: str) -> None:
-        detail = f"Fields of storage '{storage_id}' not found"
-        super().__init__(HTTPStatus.NOT_FOUND, detail)
+# Naming convention for exceptions related to the study configuration:
+#
+# | Topic         | NotFound (404)        | Duplicate (409)        | Invalid (422)        |
+# |---------------|-----------------------|------------------------|----------------------|
+# | ConfigFile    | ConfigFileNotFound    | N/A                    | InvalidConfigFile    |
+# | ConfigSection | ConfigSectionNotFound | DuplicateConfigSection | InvalidConfigSection |
+# | ConfigOption  | ConfigOptionNotFound  | DuplicateConfigOption  | InvalidConfigOption  |
+# | Matrix        | MatrixNotFound        | DuplicateMatrix        | InvalidMatrix        |
 
-    def __str__(self) -> str:
-        return self.detail
 
+THERMAL_CLUSTER = "thermal cluster"
+RENEWABLE_CLUSTER = "renewable cluster"
+SHORT_TERM_STORAGE = "short-term storage"
+
+# ============================================================
+# NotFound (404)
+# ============================================================
+
+_match_input_path = re.compile(r"input(?:/[\w*-]+)+").fullmatch
 
-class STStorageMatrixNotFoundError(HTTPException):
-    """Matrix of the short-term storage is not found"""
 
-    def __init__(self, study_id: str, area_id: str, storage_id: str, ts_name: str) -> None:
-        detail = f"Time series '{ts_name}' of storage '{storage_id}' not found"
+class ConfigFileNotFound(HTTPException):
+    """
+    Exception raised when a configuration file is not found (404 Not Found).
+
+    Notes:
+        The study ID is not provided because it is implicit.
+
+    Attributes:
+        path: Path of the missing file(s) relative to the study directory.
+        area_ids: Sequence of area IDs for which the file(s) is/are missing.
+    """
+
+    object_name = ""
+    """Name of the object that is not found: thermal, renewables, etc."""
+
+    def __init__(self, path: str, *area_ids: str):
+        assert _match_input_path(path), f"Invalid path: '{path}'"
+        self.path = path
+        self.area_ids = area_ids
+        ids = ", ".join(f"'{a}'" for a in area_ids)
+        detail = {
+            0: f"Path '{path}' not found",
+            1: f"Path '{path}' not found for area {ids}",
+            2: f"Path '{path}' not found for areas {ids}",
+        }[min(len(area_ids), 2)]
+        if self.object_name:
+            detail = f"{self.object_name.title()} {detail}"
         super().__init__(HTTPStatus.NOT_FOUND, detail)
 
     def __str__(self) -> str:
+        """Return a string representation of the exception."""
         return self.detail
 
 
-class STStorageConfigNotFoundError(HTTPException):
-    """Configuration for short-term storage is not found"""
+class ThermalClusterConfigNotFound(ConfigFileNotFound):
+    """Configuration for thermal cluster is not found (404 Not Found)"""
+
+    object_name = THERMAL_CLUSTER
+
+
+class RenewableClusterConfigNotFound(ConfigFileNotFound):
+    """Configuration for renewable cluster is not found (404 Not Found)"""
+
+    object_name = RENEWABLE_CLUSTER
+
+
+class STStorageConfigNotFound(ConfigFileNotFound):
+    """Configuration for short-term storage is not found (404 Not Found)"""
+
+    object_name = SHORT_TERM_STORAGE
+
+
+class ConfigSectionNotFound(HTTPException):
+    """
+    Exception raised when a configuration section is not found (404 Not Found).
 
-    def __init__(self, study_id: str, area_id: str) -> None:
-        detail = f"The short-term storage configuration of area '{area_id}' not found"
+    Notes:
+        The study ID is not provided because it is implicit.
+
+    Attributes:
+        path: Path of the missing file(s) relative to the study directory.
+        section_id: ID of the missing section.
+    """
+
+    object_name = ""
+    """Name of the object that is not found: thermal, renewables, etc."""
+
+    def __init__(self, path: str, section_id: str):
+        assert _match_input_path(path), f"Invalid path: '{path}'"
+        self.path = path
+        self.section_id = section_id
+        object_name = self.object_name or "section"
+        detail = f"{object_name.title()} '{section_id}' not found in '{path}'"
         super().__init__(HTTPStatus.NOT_FOUND, detail)
 
     def __str__(self) -> str:
+        """Return a string representation of the exception."""
         return self.detail
 
 
-class STStorageNotFoundError(HTTPException):
-    """Short-term storage is not found"""
+class ThermalClusterNotFound(ConfigSectionNotFound):
+    """Thermal cluster is not found (404 Not Found)"""
+
+    object_name = THERMAL_CLUSTER
+
+
+class RenewableClusterNotFound(ConfigSectionNotFound):
+    """Renewable cluster is not found (404 Not Found)"""
+
+    object_name = RENEWABLE_CLUSTER
+
+
+class STStorageNotFound(ConfigSectionNotFound):
+    """Short-term storage is not found (404 Not Found)"""
+
+    object_name = SHORT_TERM_STORAGE
 
-    def __init__(self, study_id: str, area_id: str, st_storage_id: str) -> None:
-        detail = f"Short-term storage '{st_storage_id}' not found in area '{area_id}'"
+
+class MatrixNotFound(HTTPException):
+    """
+    Exception raised when a matrix is not found (404 Not Found).
+
+    Notes:
+        The study ID is not provided because it is implicit.
+
+    Attributes:
+        path: Path of the missing file(s) relative to the study directory.
+    """
+
+    object_name = ""
+    """Name of the object that is not found: thermal, renewables, etc."""
+
+    def __init__(self, path: str):
+        assert _match_input_path(path), f"Invalid path: '{path}'"
+        self.path = path
+        detail = f"Matrix '{path}' not found"
+        if self.object_name:
+            detail = f"{self.object_name.title()} {detail}"
         super().__init__(HTTPStatus.NOT_FOUND, detail)
 
     def __str__(self) -> str:
         return self.detail
 
 
-class DuplicateSTStorageId(HTTPException):
-    """Exception raised when trying to create a short-term storage with an already existing id."""
+class ThermalClusterMatrixNotFound(MatrixNotFound):
+    """Matrix of the thermal cluster is not found (404 Not Found)"""
+
+    object_name = THERMAL_CLUSTER
+
+
+class RenewableClusterMatrixNotFound(MatrixNotFound):
+    """Matrix of the renewable cluster is not found (404 Not Found)"""
+
+    object_name = RENEWABLE_CLUSTER
 
-    def __init__(self, study_id: str, area_id: str, st_storage_id: str) -> None:
-        detail = f"Short term storage '{st_storage_id}' already exists in area '{area_id}'"
+
+class STStorageMatrixNotFound(MatrixNotFound):
+    """Matrix of the short-term storage is not found (404 Not Found)"""
+
+    object_name = SHORT_TERM_STORAGE
+
+
+# ============================================================
+# Duplicate (409)
+# ============================================================
+
+
+class DuplicateConfigSection(HTTPException):
+    """
+    Exception raised when a configuration section is duplicated (409 Conflict).
+
+    Notes:
+        The study ID is not provided because it is implicit.
+
+    Attributes:
+        area_id: ID of the area in which the section is duplicated.
+        duplicates: Sequence of duplicated IDs.
+    """
+
+    object_name = ""
+    """Name of the object that is duplicated: thermal, renewables, etc."""
+
+    def __init__(self, area_id: str, *duplicates: str):
+        self.area_id = area_id
+        self.duplicates = duplicates
+        ids = ", ".join(f"'{a}'" for a in duplicates)
+        detail = {
+            0: f"Duplicates found in '{area_id}'",
+            1: f"Duplicate found in '{area_id}': {ids}",
+            2: f"Duplicates found in '{area_id}': {ids}",
+        }[min(len(duplicates), 2)]
+        if self.object_name:
+            detail = f"{self.object_name.title()} {detail}"
         super().__init__(HTTPStatus.CONFLICT, detail)
 
     def __str__(self) -> str:
+        """Return a string representation of the exception."""
         return self.detail
 
 
-class UnknownModuleError(Exception):
-    def __init__(self, message: str) -> None:
-        super(UnknownModuleError, self).__init__(message)
+class DuplicateThermalCluster(DuplicateConfigSection):
+    """Duplicate Thermal cluster (409 Conflict)"""
+
+    object_name = THERMAL_CLUSTER
+
+
+class DuplicateRenewableCluster(DuplicateConfigSection):
+    """Duplicate Renewable cluster (409 Conflict)"""
+
+    object_name = RENEWABLE_CLUSTER
+
+
+class DuplicateSTStorage(DuplicateConfigSection):
+    """Duplicate Short-term storage (409 Conflict)"""
+
+    object_name = SHORT_TERM_STORAGE
 
 
 class StudyNotFoundError(HTTPException):
@@ -108,11 +272,6 @@ def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.LOCKED, message)
 
 
-class StudyAlreadyExistError(HTTPException):
-    def __init__(self, message: str) -> None:
-        super().__init__(HTTPStatus.CONFLICT, message)
-
-
 class StudyValidationError(HTTPException):
     def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.UNPROCESSABLE_ENTITY, message)
@@ -328,29 +487,3 @@ def __init__(self) -> None:
             HTTPStatus.BAD_REQUEST,
             "You cannot scan the default internal workspace",
         )
-
-
-class ClusterNotFound(HTTPException):
-    def __init__(self, cluster_id: str) -> None:
-        super().__init__(
-            HTTPStatus.NOT_FOUND,
-            f"Cluster: '{cluster_id}' not found",
-        )
-
-
-class ClusterConfigNotFound(HTTPException):
-    def __init__(self, area_id: str) -> None:
-        super().__init__(
-            HTTPStatus.NOT_FOUND,
-            f"Cluster configuration for area: '{area_id}' not found",
-        )
-
-
-class ClusterAlreadyExists(HTTPException):
-    """Exception raised when attempting to create a cluster with an already existing ID."""
-
-    def __init__(self, cluster_type: str, cluster_id: str) -> None:
-        super().__init__(
-            HTTPStatus.CONFLICT,
-            f"{cluster_type} cluster with ID '{cluster_id}' already exists and could not be created.",
-        )

antarest/matrixstore/service.py

@@ -86,8 +86,13 @@ def get_matrix_id(self, matrix: t.Union[t.List[t.List[float]], str]) -> str:
         Raises:
             TypeError: If the provided matrix is neither a matrix nor a link to a matrix.
         """
+        # noinspection SpellCheckingInspection
         if isinstance(matrix, str):
-            return matrix.lstrip("matrix://")
+            # str.removeprefix() is not available in Python 3.8
+            prefix = "matrix://"
+            if matrix.startswith(prefix):
+                return matrix[len(prefix) :]
+            return matrix
         elif isinstance(matrix, list):
             return self.create(matrix)
         else:

antarest/study/business/areas/renewable_management.py

@@ -3,7 +3,7 @@
 
 from pydantic import validator
 
-from antarest.core.exceptions import ClusterAlreadyExists, ClusterConfigNotFound, ClusterNotFound
+from antarest.core.exceptions import DuplicateRenewableCluster, RenewableClusterConfigNotFound, RenewableClusterNotFound
 from antarest.study.business.enum_ignore_case import EnumIgnoreCase
 from antarest.study.business.utils import AllOptionalMetaclass, camel_case_model, execute_or_add_commands
 from antarest.study.model import Study
@@ -132,15 +132,15 @@ def get_clusters(self, study: Study, area_id: str) -> t.Sequence[RenewableCluste
             List of cluster output for all clusters.
 
         Raises:
-            ClusterConfigNotFound: If the clusters configuration for the specified area is not found.
+            RenewableClusterConfigNotFound: If the clusters configuration for the specified area is not found.
         """
         file_study = self._get_file_study(study)
         path = _CLUSTERS_PATH.format(area_id=area_id)
 
         try:
             clusters = file_study.tree.get(path.split("/"), depth=3)
         except KeyError:
-            raise ClusterConfigNotFound(area_id)
+            raise RenewableClusterConfigNotFound(path, area_id)
 
         return [create_renewable_output(study.version, cluster_id, cluster) for cluster_id, cluster in clusters.items()]
 
@@ -192,14 +192,14 @@ def get_cluster(self, study: Study, area_id: str, cluster_id: str) -> RenewableC
             The cluster output representation.
 
         Raises:
-            ClusterNotFound: If the specified cluster is not found within the area.
+            RenewableClusterNotFound: If the specified cluster is not found within the area.
         """
         file_study = self._get_file_study(study)
         path = _CLUSTER_PATH.format(area_id=area_id, cluster_id=cluster_id)
         try:
             cluster = file_study.tree.get(path.split("/"), depth=1)
         except KeyError:
-            raise ClusterNotFound(cluster_id)
+            raise RenewableClusterNotFound(path, cluster_id)
         return create_renewable_output(study.version, cluster_id, cluster)
 
     def update_cluster(
@@ -222,7 +222,7 @@ def update_cluster(
             The updated cluster configuration.
 
         Raises:
-            ClusterNotFound: If the cluster to update is not found.
+            RenewableClusterNotFound: If the cluster to update is not found.
         """
 
         study_version = study.version
@@ -232,7 +232,7 @@ def update_cluster(
         try:
             values = file_study.tree.get(path.split("/"), depth=1)
         except KeyError:
-            raise ClusterNotFound(cluster_id) from None
+            raise RenewableClusterNotFound(path, cluster_id) from None
         else:
             old_config = create_renewable_config(study_version, **values)
 
@@ -298,12 +298,12 @@ def duplicate_cluster(
             The duplicated cluster configuration.
 
         Raises:
-            ClusterAlreadyExists: If a cluster with the new name already exists in the area.
+            DuplicateRenewableCluster: If a cluster with the new name already exists in the area.
         """
         new_id = transform_name_to_id(new_cluster_name, lower=False)
         lower_new_id = new_id.lower()
         if any(lower_new_id == cluster.id.lower() for cluster in self.get_clusters(study, area_id)):
-            raise ClusterAlreadyExists("Renewable", new_id)
+            raise DuplicateRenewableCluster(area_id, new_id)
 
         # Cluster duplication
         current_cluster = self.get_cluster(study, area_id, source_id)