More centralized exception handling

Author: okorach-sonar

cli/projects_cli.py

@@ -24,7 +24,6 @@
 
 """
 
-import sys
 import json
 
 from requests import RequestException

sonar/app_branches.py

@@ -24,8 +24,6 @@
 from typing import Optional
 
 import json
-from http import HTTPStatus
-from requests import RequestException
 from requests.utils import quote
 
 import sonar.logging as log
@@ -112,11 +110,7 @@ def create(cls, app: object, name: str, project_branches: list[Branch]) -> Appli
             else:  # Default main branch of project
                 params["project"].append(obj.key)
                 params["projectBranch"].append("")
-        try:
-            app.endpoint.post(ApplicationBranch.API[c.CREATE], params=params)
-        except (ConnectionError, RequestException) as e:
-            utilities.handle_error(e, f"creating branch {name} of {str(app)}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
-            raise exceptions.ObjectAlreadyExists(f"{str(app)} branch '{name}", e.response.text)
+        app.endpoint.post(ApplicationBranch.API[c.CREATE], params=params)
         return ApplicationBranch(app=app, name=name, project_branches=project_branches)
 
     @classmethod

sonar/applications.py

@@ -128,11 +128,7 @@ def create(cls, endpoint: pf.Platform, key: str, name: str) -> Application:
         :rtype: Application
         """
         check_supported(endpoint)
-        try:
-            endpoint.post(Application.API["CREATE"], params={"key": key, "name": name})
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"creating application {key}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
-            raise exceptions.ObjectAlreadyExists(key, e.response.text)
+        endpoint.post(Application.API["CREATE"], params={"key": key, "name": name})
         log.info("Creating object")
         return Application(endpoint=endpoint, key=key, name=name)
 

sonar/branches.py

@@ -26,7 +26,6 @@
 import json
 import re
 from urllib.parse import unquote
-from requests import HTTPError, RequestException
 import requests.utils
 
 from sonar import platform
@@ -173,9 +172,8 @@ def delete(self) -> bool:
         """
         try:
             return super().delete()
-        except (ConnectionError, RequestException) as e:
-            if isinstance(e, HTTPError) and e.response.status_code == HTTPStatus.BAD_REQUEST:
-                log.warning("Can't delete %s, it's the main branch", str(self))
+        except exceptions.SonarException as e:
+            log.warning(e.message)
             return False
 
     def get(self, api: str, params: types.ApiParams = None, data: str = None, mute: tuple[HTTPStatus] = (), **kwargs) -> requests.Response:

sonar/devops.py

@@ -22,11 +22,8 @@
 
 from __future__ import annotations
 from typing import Optional, Union
-from http import HTTPStatus
 import json
 
-from requests import RequestException
-
 import sonar.logging as log
 from sonar.util import types, cache
 from sonar import platform
@@ -109,11 +106,10 @@ def create(cls, endpoint: platform.Platform, key: str, plt_type: str, url_or_wor
             elif plt_type == "bitbucketcloud":
                 params.update({"clientSecret": _TO_BE_SET, "clientId": _TO_BE_SET, "workspace": url_or_workspace})
                 endpoint.post(_CREATE_API_BBCLOUD, params=params)
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"creating devops platform {key}/{plt_type}/{url_or_workspace}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
+        except exceptions.SonarException as e:
             if endpoint.edition() in (c.CE, c.DE):
                 log.warning("Can't set DevOps platform '%s', don't you have more that 1 of that type?", key)
-            raise exceptions.UnsupportedOperation(f"Can't set DevOps platform '{key}', don't you have more that 1 of that type?")
+            raise exceptions.UnsupportedOperation(e.message) from e
         o = DevopsPlatform(endpoint=endpoint, key=key, platform_type=plt_type)
         o.refresh()
         return o
@@ -185,8 +181,7 @@ def update(self, **kwargs) -> bool:
             ok = self.post(f"alm_settings/update_{alm_type}", params=params).ok
             self.url = kwargs["url"]
             self._specific = {k: v for k, v in params.items() if k not in ("key", "url")}
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"updating devops platform {self.key}/{alm_type}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
+        except exceptions.SonarException:
             ok = False
         return ok
 

sonar/findings.py

@@ -24,8 +24,6 @@
 from datetime import datetime
 from typing import Optional
 import re
-from http import HTTPStatus
-from requests import RequestException
 import Levenshtein
 
 import sonar.logging as log
@@ -460,13 +458,10 @@ def search_siblings(
     def do_transition(self, transition: str) -> bool:
         try:
             return self.post("issues/do_transition", {"issue": self.key, "transition": transition}).ok
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"applying transition {transition}", catch_http_statuses=(HTTPStatus.BAD_REQUEST, HTTPStatus.NOT_FOUND))
         except exceptions.SonarException as e:
             if re.match(r"Transition from state [A-Za-z]+ does not exist", e.message):
                 raise exceptions.UnsupportedOperation(e.message) from e
             raise
-        return False
 
     def get_branch_and_pr(self, data: types.ApiPayload) -> tuple[Optional[str], Optional[str]]:
         """

sonar/groups.py

@@ -26,9 +26,6 @@
 
 from typing import Optional
 
-from http import HTTPStatus
-from requests import RequestException
-
 import sonar.logging as log
 import sonar.platform as pf
 import sonar.sqobject as sq
@@ -115,11 +112,7 @@ def create(cls, endpoint: pf.Platform, name: str, description: str = None) -> Gr
         :return: The group object
         """
         log.debug("Creating group '%s'", name)
-        try:
-            data = json.loads(endpoint.post(Group.api_for(c.CREATE, endpoint), params={"name": name, "description": description}).text)
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"creating group '{name}'", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
-            raise exceptions.ObjectAlreadyExists(name, util.sonar_error(e.response))
+        data = json.loads(endpoint.post(Group.api_for(c.CREATE, endpoint), params={"name": name, "description": description}).text)
         o = cls.read(endpoint=endpoint, name=name)
         o.sq_json.update(data)
         return o

sonar/hotspots.py

@@ -25,7 +25,6 @@
 from datetime import datetime
 from typing import Optional
 from http import HTTPStatus
-from requests import RequestException
 import requests.utils
 
 import sonar.logging as log
@@ -36,6 +35,7 @@
 
 from sonar import users
 from sonar import findings, rules, changelog
+from sonar import exceptions
 
 PROJECT_FILTER = "project"
 PROJECT_FILTER_OLD = "projectKey"
@@ -152,19 +152,17 @@ def refresh(self) -> bool:
                     self.rule = d["rule"]["key"]
                 self.assignee = d.get("assignee", None)
             return resp.ok
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, "refreshing hotspot", catch_all=True)
+        except exceptions.SonarException:
             return False
 
     def __mark_as(self, resolution: Optional[str], comment: Optional[str] = None, status: str = "REVIEWED") -> bool:
         try:
             params = util.remove_nones({"hotspot": self.key, "status": status, "resolution": resolution, "commemt": comment})
-            r = self.post("hotspots/change_status", params=params)
-        except (ConnectionError, requests.RequestException) as e:
-            util.handle_error(e, f"marking hotspot as {status}/{resolution}", catch_all=True)
+            ok = self.post("hotspots/change_status", params=params).ok
+            self.refresh()
+            return ok
+        except exceptions.SonarException:
             return False
-        self.refresh()
-        return r.ok
 
     def mark_as_safe(self) -> bool:
         """Marks a hotspot as safe
@@ -215,8 +213,7 @@ def add_comment(self, comment: str) -> bool:
         """
         try:
             return self.post("hotspots/add_comment", params={"hotspot": self.key, "comment": comment}).ok
-        except (ConnectionError, requests.RequestException) as e:
-            util.handle_error(e, "adding comment to hotspot", catch_all=True)
+        except exceptions.SonarException:
             return False
 
     def assign(self, assignee: Optional[str], comment: Optional[str] = None) -> bool:
@@ -231,13 +228,12 @@ def assign(self, assignee: Optional[str], comment: Optional[str] = None) -> bool
                 log.debug("Unassigning %s", str(self))
             else:
                 log.debug("Assigning %s to '%s'", str(self), str(assignee))
-            r = self.post("hotspots/assign", util.remove_nones({"hotspot": self.key, "assignee": assignee, "comment": comment}))
-            if r.ok:
+            ok = self.post("hotspots/assign", util.remove_nones({"hotspot": self.key, "assignee": assignee, "comment": comment})).ok
+            if ok:
                 self.assignee = assignee
-        except (ConnectionError, requests.RequestException) as e:
-            util.handle_error(e, "assigning/unassigning hotspot", catch_all=True)
+            return ok
+        except exceptions.SonarException:
             return False
-        return r.ok
 
     def unassign(self, comment: Optional[str] = None) -> bool:
         """Unassigns a hotspot (and optionally comment)
@@ -421,8 +417,7 @@ def search(endpoint: pf.Platform, filters: types.ApiParams = None) -> dict[str,
             try:
                 data = json.loads(endpoint.get(Hotspot.API[c.SEARCH], params=inline_filters, mute=(HTTPStatus.NOT_FOUND,)).text)
                 nbr_hotspots = util.nbr_total_elements(data)
-            except (ConnectionError, RequestException) as e:
-                util.handle_error(e, "searching hotspots", catch_all=True)
+            except exceptions.SonarException:
                 nbr_hotspots = 0
                 return {}
             nbr_pages = util.nbr_pages(data)

sonar/issues.py

@@ -293,11 +293,9 @@ def add_comment(self, comment: str) -> bool:
         """
         log.debug("Adding comment '%s' to %s", comment, str(self))
         try:
-            r = self.post("issues/add_comment", {"issue": self.key, "text": comment})
-        except requests.RequestException as e:
-            util.handle_error(e, "adding comment", catch_all=True)
+            return self.post("issues/add_comment", {"issue": self.key, "text": comment}).ok
+        except exceptions.SonarException:
             return False
-        return r.ok
 
     def __set_severity(self, **params) -> bool:
         log.debug("Changing severity of %s from '%s' to '%s'", str(self), self.severity, str(params))
@@ -343,13 +341,11 @@ def assign(self, assignee: Optional[str] = None) -> bool:
         try:
             params = util.remove_nones({"issue": self.key, "assignee": assignee})
             log.debug("Assigning %s to '%s'", str(self), str(assignee))
-            r = self.post("issues/assign", params)
-            if r.ok:
+            if ok := self.post("issues/assign", params).ok:
                 self.assignee = assignee
-        except (ConnectionError, requests.RequestException) as e:
-            util.handle_error(e, "assigning issue", catch_all=True)
+            return ok
+        except exceptions.SonarException:
             return False
-        return r.ok
 
     def get_tags(self, **kwargs) -> list[str]:
         """Returns issues tags"""
@@ -395,13 +391,11 @@ def set_type(self, new_type: str) -> bool:
             raise exceptions.UnsupportedOperation("Changing issue type is not supported in MQR mode")
         log.debug("Changing type of issue %s from %s to %s", self.key, self.type, new_type)
         try:
-            r = self.post("issues/set_type", {"issue": self.key, "type": new_type})
-            if r.ok:
+            if ok := self.post("issues/set_type", {"issue": self.key, "type": new_type}).ok:
                 self.type = new_type
-        except (ConnectionError, requests.RequestException) as e:
-            util.handle_error(e, "setting issue type", catch_all=True)
+            return ok
+        except exceptions.SonarException:
             return False
-        return r.ok
 
     def is_wont_fix(self) -> bool:
         """

sonar/permissions/permission_templates.py

@@ -24,11 +24,10 @@
 
 import json
 import re
-from requests import RequestException
 
 import sonar.logging as log
 from sonar.util import types, cache
-from sonar import sqobject, utilities
+from sonar import sqobject, utilities, exceptions
 from sonar.permissions import template_permissions
 import sonar.platform as pf
 from sonar.audit.rules import get_rule, RuleId
@@ -136,7 +135,7 @@ def permissions(self) -> template_permissions.TemplatePermissions:
             self._permissions = template_permissions.TemplatePermissions(self)
         return self._permissions
 
-    def set_as_default(self, what_list: list[str]) -> None:
+    def set_as_default(self, what_list: list[str]) -> bool:
         """Sets a permission template as default for projects or apps or portfolios"""
         log.debug("Setting %s as default for %s", str(self), str(what_list))
         ed = self.endpoint.edition()
@@ -146,9 +145,9 @@ def set_as_default(self, what_list: list[str]) -> None:
                 log.warning("Can't set permission template as default for %s on a %s edition", qual, ed)
                 continue
             try:
-                self.post("permissions/set_default_template", params={"templateId": self.key, "qualifier": qual})
-            except (ConnectionError, RequestException) as e:
-                utilities.handle_error(e, f"setting {str(self)} as default")
+                return self.post("permissions/set_default_template", params={"templateId": self.key, "qualifier": qual}).ok
+            except exceptions.SonarException:
+                return False
 
     def set_pattern(self, pattern: str) -> PermissionTemplate:
         """Sets a permission template pattern"""