feat(delete-apis): simplify error handling for deletions with binding…

… constraints

antarest/core/exceptions.py

@@ -4,8 +4,6 @@
 
 from fastapi.exceptions import HTTPException
 
-MAX_BINDING_CONSTRAINTS_TO_DISPLAY = 10
-
 
 class ShouldNotHappenException(Exception):
     pass
@@ -327,43 +325,33 @@ def __init__(self, is_variant: bool) -> None:
             super().__init__(HTTPStatus.EXPECTATION_FAILED, "Upgrade not supported for parent of variants")
 
 
-class AreaDeletionNotAllowed(HTTPException):
-    def __init__(self, uuid: str, binding_constraints_ids: t.List[str]) -> None:
-        first_bcs_ids = ""
-        for i, bc in enumerate(binding_constraints_ids[:MAX_BINDING_CONSTRAINTS_TO_DISPLAY]):
-            first_bcs_ids += f"{i+1}- {bc}\n"
-        message = "Area is referenced in the following binding constraints:\n" + first_bcs_ids
-        msg = f"Area {uuid} is not allowed to be deleted\n{message}"
-        super().__init__(
-            HTTPStatus.FORBIDDEN,
-            msg,
-        )
-
+class BindingConstraintDeletionNotAllowed(HTTPException):
+    """
+    Exception raised when a binding constraint is not allowed to be deleted because it references
+    other objects: areas, links or thermal clusters.
+    """
 
-class LinkDeletionNotAllowed(HTTPException):
-    def __init__(self, uuid: str, binding_constraints_ids: t.List[str]) -> None:
-        first_bcs_ids = ""
-        for i, bc in enumerate(binding_constraints_ids[:MAX_BINDING_CONSTRAINTS_TO_DISPLAY]):
-            first_bcs_ids += f"{i+1}- {bc}\n"
-        message = "Link is referenced in the following binding constraints:\n" + first_bcs_ids
-        msg = f"Link {uuid} is not allowed to be deleted\n{message}"
-        super().__init__(
-            HTTPStatus.FORBIDDEN,
-            msg,
+    def __init__(self, object_id: str, binding_ids: t.Sequence[str], *, object_type: str) -> None:
+        """
+        Initialize the exception.
+
+        Args:
+            object_id: ID of the object that is not allowed to be deleted.
+            binding_ids: Binding constraints IDs that reference the object.
+            object_type: Type of the object that is not allowed to be deleted: area, link or thermal cluster.
+        """
+        max_count = 10
+        first_bcs_ids = ",\n".join(f"{i}- '{bc}'" for i, bc in enumerate(binding_ids[:max_count], 1))
+        and_more = f",\nand {len(binding_ids) - max_count} more..." if len(binding_ids) > max_count else "."
+        message = (
+            f"{object_type} '{object_id}' is not allowed to be deleted, because it is referenced"
+            f" in the following binding constraints:\n{first_bcs_ids}{and_more}"
         )
+        super().__init__(HTTPStatus.FORBIDDEN, message)
 
-
-class ClusterDeletionNotAllowed(HTTPException):
-    def __init__(self, uuid: str, binding_constraints_ids: t.List[str]) -> None:
-        first_bcs_ids = ""
-        for i, bc in enumerate(binding_constraints_ids[:MAX_BINDING_CONSTRAINTS_TO_DISPLAY]):
-            first_bcs_ids += f"{i+1}- {bc}\n"
-        message = "Cluster is referenced in the following binding constraints:\n" + first_bcs_ids
-        msg = f"Cluster {uuid} is not allowed to be deleted\n{message}"
-        super().__init__(
-            HTTPStatus.FORBIDDEN,
-            msg,
-        )
+    def __str__(self) -> str:
+        """Return a string representation of the exception."""
+        return self.detail
 
 
 class UnsupportedStudyVersion(HTTPException):

antarest/study/business/binding_constraint_management.py

@@ -501,13 +501,12 @@ def terms_to_coeffs(terms: t.Sequence[ConstraintTerm]) -> t.Dict[str, t.List[flo
         :return: A dictionary of term IDs mapped to a list of their coefficients.
         """
         coeffs = {}
-        if terms is not None:
-            for term in terms:
-                if term.id and term.weight is not None:
-                    coeffs[term.id] = [term.weight]
-                    if term.offset:
-                        coeffs[term.id].append(term.offset)
-            return coeffs
+        for term in terms:
+            if term.id and term.weight is not None:
+                coeffs[term.id] = [term.weight]
+                if term.offset:
+                    coeffs[term.id].append(term.offset)
+        return coeffs
 
     def get_binding_constraint(self, study: Study, bc_id: str) -> ConstraintOutput:
         """

antarest/study/service.py

@@ -20,12 +20,10 @@
 
 from antarest.core.config import Config
 from antarest.core.exceptions import (
-    AreaDeletionNotAllowed,
     BadEditInstructionException,
-    ClusterDeletionNotAllowed,
+    BindingConstraintDeletionNotAllowed,
     CommandApplicationError,
     IncorrectPathError,
-    LinkDeletionNotAllowed,
     NotAManagedStudyException,
     StudyDeletionNotAllowed,
     StudyNotFoundError,
@@ -689,7 +687,7 @@ def create_study(
             id=sid,
             name=study_name,
             workspace=DEFAULT_WORKSPACE_NAME,
-            path=study_path,
+            path=str(study_path),
             created_at=datetime.utcnow(),
             updated_at=datetime.utcnow(),
             version=version or NEW_DEFAULT_STUDY_VERSION,
@@ -1268,17 +1266,18 @@ def export_task(notifier: TaskUpdateNotifier) -> TaskResult:
                 stopwatch.log_elapsed(
                     lambda x: logger.info(f"Study {study_id} filtered output {output_id} exported in {x}s")
                 )
-                return FileResponse(
-                    tmp_export_file,
-                    headers=(
-                        {"Content-Disposition": "inline"}
-                        if filetype == ExportFormat.JSON
-                        else {
-                            "Content-Disposition": f'attachment; filename="output-{output_id}.{"tar.gz" if filetype == ExportFormat.TAR_GZ else "zip"}'
-                        }
-                    ),
-                    media_type=filetype,
-                )
+
+                if filetype == ExportFormat.JSON:
+                    headers = {"Content-Disposition": "inline"}
+                elif filetype == ExportFormat.TAR_GZ:
+                    headers = {"Content-Disposition": f'attachment; filename="output-{output_id}.tar.gz'}
+                elif filetype == ExportFormat.ZIP:
+                    headers = {"Content-Disposition": f'attachment; filename="output-{output_id}.zip'}
+                else:  # pragma: no cover
+                    raise NotImplementedError(f"Export format {filetype} is not supported")
+
+                return FileResponse(tmp_export_file, headers=headers, media_type=filetype)
+
             else:
                 json_response = json.dumps(
                     matrix.dict(),
@@ -1317,26 +1316,20 @@ def set_sim_reference(
         params: RequestParameters,
     ) -> None:
         """
-        Set simulation as the reference output
+        Set simulation as the reference output.
+
         Args:
-            study_id: study Id
-            output_id: output id
-            status: state of the reference status
+            study_id: study ID.
+            output_id: The ID of the output to set as reference.
+            status: state of the reference status.
             params: request parameters
-
-        Returns: None
-
         """
         study = self.get_study(study_id)
         assert_permission(params.user, study, StudyPermissionType.WRITE)
         self._assert_study_unarchived(study)
 
         logger.info(
-            "output %s set by user %s as reference (%b) for study %s",
-            output_id,
-            params.get_user_id(),
-            status,
-            study_id,
+            f"output {output_id} set by user {params.get_user_id()} as reference ({status}) for study {study_id}"
         )
 
         self.storage_service.get_storage(study).set_reference_output(study, output_id, status)
@@ -1858,14 +1851,27 @@ def update_thermal_cluster_metadata(
         return self.areas.update_thermal_cluster_metadata(study, area_id, clusters_metadata)
 
     def delete_area(self, uuid: str, area_id: str, params: RequestParameters) -> None:
+        """
+        Delete area from study if it is not referenced by a binding constraint,
+        otherwise raise an HTTP 403 Forbidden error.
+
+        Args:
+            uuid: The study ID.
+            area_id: The area ID to delete.
+            params: The request parameters used to check user permissions.
+
+        Raises:
+            BindingConstraintDeletionNotAllowed: If the area is referenced by a binding constraint.
+        """
         study = self.get_study(uuid)
         assert_permission(params.user, study, StudyPermissionType.WRITE)
         self._assert_study_unarchived(study)
         referencing_binding_constraints = self.binding_constraint_manager.get_binding_constraints(
             study, ConstraintFilters(area_name=area_id)
         )
         if referencing_binding_constraints:
-            raise AreaDeletionNotAllowed(area_id, [bc.id for bc in referencing_binding_constraints])
+            binding_ids = [bc.id for bc in referencing_binding_constraints]
+            raise BindingConstraintDeletionNotAllowed(area_id, binding_ids, object_type="Area")
         self.areas.delete_area(study, area_id)
         self.event_bus.push(
             Event(
@@ -1882,6 +1888,19 @@ def delete_link(
         area_to: str,
         params: RequestParameters,
     ) -> None:
+        """
+        Delete link from study if it is not referenced by a binding constraint,
+        otherwise raise an HTTP 403 Forbidden error.
+
+        Args:
+            uuid: The study ID.
+            area_from: The area from which the link starts.
+            area_to: The area to which the link ends.
+            params: The request parameters used to check user permissions.
+
+        Raises:
+            BindingConstraintDeletionNotAllowed: If the link is referenced by a binding constraint.
+        """
         study = self.get_study(uuid)
         assert_permission(params.user, study, StudyPermissionType.WRITE)
         self._assert_study_unarchived(study)
@@ -1890,7 +1909,8 @@ def delete_link(
             study, ConstraintFilters(link_id=link_id)
         )
         if referencing_binding_constraints:
-            raise LinkDeletionNotAllowed(link_id, [bc.id for bc in referencing_binding_constraints])
+            binding_ids = [bc.id for bc in referencing_binding_constraints]
+            raise BindingConstraintDeletionNotAllowed(link_id, binding_ids, object_type="Link")
         self.links.delete_link(study, area_from, area_to)
         self.event_bus.push(
             Event(
@@ -2537,22 +2557,21 @@ def asserts_no_thermal_in_binding_constraints(
         self, study: Study, area_id: str, cluster_ids: t.Sequence[str]
     ) -> None:
         """
-        Check that no cluster is referenced in a binding constraint otherwise raise an ClusterDeletionNotAllowed
-        Exception
+        Check that no cluster is referenced in a binding constraint, otherwise raise an HTTP 403 Forbidden error.
 
         Args:
             study: input study for which an update is to be committed
             area_id: area ID to be checked
             cluster_ids: IDs of the thermal clusters to be checked
 
-        Returns: None
-
-        Raises: ClusterDeletionNotAllowed if a cluster is referenced in a binding constraint
+        Raises:
+            BindingConstraintDeletionNotAllowed: if a cluster is referenced in a binding constraint
         """
 
         for cluster_id in cluster_ids:
-            referencing_binding_constraints = self.binding_constraint_manager.get_binding_constraints(
+            ref_bcs = self.binding_constraint_manager.get_binding_constraints(
                 study, ConstraintFilters(cluster_id=f"{area_id}.{cluster_id}")
             )
-            if referencing_binding_constraints:
-                raise ClusterDeletionNotAllowed(cluster_id, [bc.id for bc in referencing_binding_constraints])
+            if ref_bcs:
+                binding_ids = [bc.id for bc in ref_bcs]
+                raise BindingConstraintDeletionNotAllowed(cluster_id, binding_ids, object_type="Cluster")

tests/core/test_exceptions.py

@@ -0,0 +1,25 @@
+from antarest.core.exceptions import BindingConstraintDeletionNotAllowed
+
+
+class TestBindingConstraintDeletionNotAllowed:
+    def test_few_binding_constraints(self) -> None:
+        object_id = "france"
+        binding_ids = ["bc1", "bc2"]
+        object_type = "Area"
+        exception = BindingConstraintDeletionNotAllowed(object_id, binding_ids, object_type=object_type)
+        message = str(exception)
+        assert f"{object_type} '{object_id}'" in message
+        assert "bc1" in message
+        assert "bc2" in message
+        assert "more..." not in message
+
+    def test_many_binding_constraints(self) -> None:
+        object_id = "france"
+        binding_ids = [f"bc{i}" for i in range(1, 50)]
+        object_type = "Area"
+        exception = BindingConstraintDeletionNotAllowed(object_id, binding_ids, object_type=object_type)
+        message = str(exception)
+        assert f"{object_type} '{object_id}'" in message
+        assert "bc1" in message
+        assert "bc2" in message
+        assert "more..." in message

tests/integration/study_data_blueprint/test_thermal.py

@@ -656,7 +656,7 @@ def test_lifecycle(
         assert res.status_code == 403, res.json()
         description = res.json()["description"]
         assert all([elm in description for elm in [fr_gas_conventional, "binding constraint"]])
-        assert res.json()["exception"] == "ClusterDeletionNotAllowed"
+        assert res.json()["exception"] == "BindingConstraintDeletionNotAllowed"
 
         # delete the binding constraint
         res = client.delete(