Implement custom exceptions for BCB API errors and update error handling in currency and OData modules

Author: wilsonfreitas

bcb/__init__.py

@@ -1,3 +1,4 @@
+from .exceptions import BCBError, BCBAPIError, CurrencyNotFoundError, SGSError, ODataError
 from .odata.api import (
     ODataAPI,
     Expectativas,

bcb/currency.py

@@ -9,6 +9,7 @@
 import requests
 from lxml import html
 
+from .exceptions import BCBAPIError, CurrencyNotFoundError
 from .utils import Date, DateInput
 
 """
@@ -37,7 +38,7 @@ def _currency_id_list() -> pd.DataFrame:
         res = requests.get(url1)
         if res.status_code != 200:
             msg = f"BCB API Request error, status code = {res.status_code}"
-            raise Exception(msg)
+            raise BCBAPIError(msg, res.status_code)
 
         doc = html.parse(BytesIO(res.content)).getroot()
         xpath = "//select[@name='ChkMoeda']/option"
@@ -183,6 +184,6 @@ def get(
             elif groupby == "side":
                 return df.reorder_levels([1, 0], axis=1).sort_index(axis=1)
         else:
-            raise Exception(f"Unknown side value, use: bid, ask, both")
+            raise ValueError("Unknown side value, use: bid, ask, both")
     else:
-        raise Exception(f"Currency not found: {symbols}")
+        raise CurrencyNotFoundError(f"Currency not found: {symbols}")

bcb/exceptions.py

@@ -0,0 +1,22 @@
+class BCBError(Exception):
+    """Base exception for all python-bcb errors."""
+
+
+class BCBAPIError(BCBError):
+    """HTTP or API-level error from BCB."""
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class CurrencyNotFoundError(BCBError):
+    """Raised when a requested currency symbol is not found."""
+
+
+class SGSError(BCBError):
+    """Raised for SGS-specific API errors."""
+
+
+class ODataError(BCBError):
+    """Raised for OData query/metadata errors."""

bcb/odata/framework.py

@@ -5,6 +5,8 @@
 import httpx
 from urllib.parse import quote
 
+from bcb.exceptions import ODataError
+
 
 # Edm.Boolean
 # Edm.Byte
@@ -358,7 +360,7 @@ def __getitem__(self, item):
             if fi is not None:
                 return fi
             else:
-                raise ValueError("Invalid name: " + item)
+                raise ODataError("Invalid name: " + item)
         else:
             return es
 
@@ -427,7 +429,7 @@ def parameters(self, **kwargs):
             if arg in self.function_parameters:
                 self.function_parameters[arg] = kwargs[arg]
             else:
-                raise ValueError(f"Unknown parameter: {arg}")
+                raise ODataError(f"Unknown parameter: {arg}")
         return self
 
     def filter(self, *args):
@@ -489,7 +491,7 @@ def text(self):
             for p in self.entity.function.parameters:
                 val = self.function_parameters[p.name]
                 if p.required and val is None:
-                    raise ValueError("Parameter not set: " + p.name)
+                    raise ODataError("Parameter not set: " + p.name)
                 params["@" + p.name] = p.format(val)
         qs = "&".join([f"{quote(k)}={quote(str(v))}" for k, v in params.items()])
         headers = {"OData-Version": "4.0", "OData-MaxVersion": "4.0"}

bcb/sgs/__init__.py

@@ -5,6 +5,7 @@
 import pandas as pd
 import requests
 
+from bcb.exceptions import SGSError
 from bcb.utils import Date, DateInput
 
 """
@@ -181,8 +182,8 @@ def get_json(code: int, start: Optional[DateInput] = None, end: Optional[DateInp
         except Exception:
             res_json = {}
         if "error" in res_json:
-            raise Exception(f"BCB error: {res_json['error']}")
+            raise SGSError(f"BCB error: {res_json['error']}")
         elif "erro" in res_json:
-            raise Exception(f"BCB error: {res_json['erro']['detail']}")
-        raise Exception(f"Download error: code = {code}")
+            raise SGSError(f"BCB error: {res_json['erro']['detail']}")
+        raise SGSError(f"Download error: code = {code}")
     return res.text

bcb/utils.py

@@ -29,7 +29,7 @@ def __init__(self, d: DateInput, format: str = "%Y-%m-%d") -> None:
         elif isinstance(d, date):
             pass
         else:
-            raise ValueError()
+            raise ValueError(f"Unsupported date type: {type(d).__name__}")
         self.date: date = d
 
     def format(self, fmts: str = "%Y-%m-%d") -> str:

tests/test_currency.py

@@ -1,7 +1,9 @@
 from datetime import datetime, date
 import pandas as pd
+import pytest
 from pytest import mark
 from bcb import currency
+from bcb.exceptions import CurrencyNotFoundError
 
 
 def test_currency_id():
@@ -18,10 +20,10 @@ def test_currency_get_symbol():
     assert x is None
     x = currency.get("USD", start_date, end_date)
     assert isinstance(x, pd.DataFrame)
-    x = currency.get("ZAR", start_date, end_date)
-    assert x is None
-    x = currency.get(["ZAR", "ZZ1"], start_date, end_date)
-    assert x is None
+    with pytest.raises(CurrencyNotFoundError):
+        currency.get("ZAR", start_date, end_date)
+    with pytest.raises(CurrencyNotFoundError):
+        currency.get(["ZAR", "ZZ1"], start_date, end_date)
 
 
 def test_get_valid_currency_list():

tests/test_expectativas.py

@@ -12,9 +12,12 @@ def endpoints():
 
 
 def test_expectativas_date_format(endpoints):
+    date_columns = {"Data", "dataHoraCotacao", "InicioPeriodo", "FimPeriodo", "DataVigencia"}
     for endpoint in endpoints:
         query = endpoint.query().limit(1)
         data = query.collect()
         assert isinstance(data, pd.DataFrame)
         assert data.shape[0] == 1
-        assert isinstance(data["Data"].iloc[0], datetime)
+        for col in date_columns:
+            if col in data.columns:
+                assert isinstance(data[col].iloc[0], datetime)