fix(errors): standardize error handling with custom exception classes

src/lighthouseweb3/functions/axios.py

@@ -4,6 +4,7 @@
 import json
 import requests as req
 from . import utils
+from .exceptions import LighthouseAPIError
 
 
 class Axios:
@@ -13,64 +14,59 @@ def __init__(self, url: str):
         self.url = url
 
     def parse_url_query(self, query):
-        try:
-            if query is not None and isinstance(query, dict):
-                for key, value in query.items():
-                    self.url += f"&{key}={value}"
-        except Exception as e:
-            raise e
+        if query is not None and isinstance(query, dict):
+            for key, value in query.items():
+                self.url += f"&{key}={value}"
 
-    def get(self, headers = None, **kwargs) :
+    def get(self, headers=None, **kwargs):
         try:
             self.parse_url_query(kwargs.get("query", None))
             r = req.get(self.url, headers=headers)
             r.raise_for_status()
             return r.json()
-        except Exception as e:
-            raise e
+        except req.exceptions.HTTPError as e:
+            raise LighthouseAPIError(str(e)) from e
 
-    def post(
-        self, body=None, headers= None, **kwargs
-    ):
+    def post(self, body=None, headers=None, **kwargs):
         try:
             self.parse_url_query(kwargs.get("query", None))
             r = req.post(self.url, data=body, headers=headers)
             r.raise_for_status()
             return r.json()
-        except Exception as e:
-            raise e
+        except req.exceptions.HTTPError as e:
+            raise LighthouseAPIError(str(e)) from e
 
-    def post_files(
-        self, file, headers = None, **kwargs
-    ) :
+    def post_files(self, file, headers=None, **kwargs):
+        files = None
         try:
             self.parse_url_query(kwargs.get("query", None))
             files = utils.read_files_for_upload(file)
             r = req.post(self.url, headers=headers, files=files)
             r.raise_for_status()
-            utils.close_files_after_upload(files)
             try:
                 return r.json()
             except Exception:
                 temp = r.text.split("\n")
                 return json.loads(temp[len(temp) - 2])
-        except Exception as e:
-            utils.close_files_after_upload(files)
-            raise e
+        except req.exceptions.RequestException as e:
+            raise LighthouseAPIError(str(e)) from e
+        finally:
+            if files is not None:
+                utils.close_files_after_upload(files)
 
-    def post_blob(
-        self, file: BufferedReader, filename: str, headers = None, **kwargs
-    ) :
+    def post_blob(self, file: BufferedReader, filename: str, headers=None, **kwargs):
         try:
             self.parse_url_query(kwargs.get("query", None))
-            files = [(
-                "file",
+            files = [
                 (
-                    utils.extract_file_name(filename),
-                    file.read(),
-                    "application/octet-stream",
+                    "file",
+                    (
+                        utils.extract_file_name(filename),
+                        file.read(),
+                        "application/octet-stream",
+                    ),
                 ),
-            ),]
+            ]
             r = req.post(self.url, headers=headers, files=files)
             r.raise_for_status()
             file.close()
@@ -79,6 +75,6 @@ def post_blob(
             except Exception:
                 temp = r.text.split("\n")
                 return json.loads(temp[len(temp) - 2])
-        except Exception as e:
+        except req.exceptions.HTTPError as e:
             file.close()
-            raise e
+            raise LighthouseAPIError(str(e)) from e

src/lighthouseweb3/functions/deal_status.py

@@ -1,5 +1,6 @@
 import requests
 from .config import Config
+from .exceptions import LighthouseAPIError
 
 
 def get_deal_status(cid: str):
@@ -9,4 +10,4 @@ def get_deal_status(cid: str):
         response.raise_for_status()
         return response.json()
     except requests.HTTPError as error:
-        raise Exception(error.response.text)
+        raise LighthouseAPIError(error.response.text) from error

src/lighthouseweb3/functions/exceptions.py

@@ -0,0 +1,19 @@
+"""Custom exceptions for Lighthouse Web3 SDK."""
+
+
+class LighthouseError(Exception):
+    """Base exception class for all Lighthouse SDK errors."""
+
+    pass
+
+
+class LighthouseAPIError(LighthouseError):
+    """Exception raised for HTTP/API errors from Lighthouse services."""
+
+    pass
+
+
+class LighthouseUploadError(LighthouseError):
+    """Exception raised for upload-related errors."""
+
+    pass

src/lighthouseweb3/functions/upload.py

@@ -5,6 +5,7 @@
 from .axios import Axios
 from .utils import is_dir, walk_dir_tree, extract_file_name, NamedBufferedReader
 from .config import Config
+from .exceptions import LighthouseError, LighthouseUploadError
 
 
 def upload(source, token: str, tag: str = ""):
@@ -25,7 +26,7 @@ def upload(source, token: str, tag: str = ""):
         axios = Axios(Config.lighthouse_node + "/api/v0/add")
         # create list of files to upload
 
-        if (isinstance(source, str)):
+        if isinstance(source, str):
             file_dict = {}
 
             # check if source is a directory
@@ -45,18 +46,20 @@ def upload(source, token: str, tag: str = ""):
 
         if len(tag):
             _axios = Axios(Config.lighthouse_api + "/api/user/create_tag")
-            data = _axios.post({
-                "tag": tag,
-                "cid": hashData.get("Hash")
-            }, {
-                "Authorization": f"Bearer {token}", })
+            data = _axios.post(
+                {"tag": tag, "cid": hashData.get("Hash")},
+                {
+                    "Authorization": f"Bearer {token}",
+                },
+            )
         return {"data": hashData}
+    except LighthouseError:
+        raise
     except Exception as e:
-        print(e)
-        raise e
+        raise LighthouseUploadError(str(e)) from e
 
 
-def uploadBlob(source:  BufferedReader, filename: str, token: str, tag: str = ""):
+def uploadBlob(source: BufferedReader, filename: str, token: str, tag: str = ""):
     """
     Upload a Buffer or readable Object
     @params {source}: str, path to file or directory
@@ -77,12 +80,14 @@ def uploadBlob(source:  BufferedReader, filename: str, token: str, tag: str = ""
         hashData = axios.post_blob(source, filename, headers)
         if len(tag):
             _axios = Axios(Config.lighthouse_api + "/api/user/create_tag")
-            data = _axios.post({
-                "tag": tag,
-                "cid": hashData.get("Hash")
-            }, {
-                "Authorization": f"Bearer {token}", })
+            data = _axios.post(
+                {"tag": tag, "cid": hashData.get("Hash")},
+                {
+                    "Authorization": f"Bearer {token}",
+                },
+            )
         return {"data": hashData}
+    except LighthouseError:
+        raise
     except Exception as e:
-        print(e)
-        raise e
+        raise LighthouseUploadError(str(e)) from e