Add detailed information for overflow error while data conversion

simplesqlite/converter.py

@@ -3,15 +3,24 @@
 """
 
 from decimal import Decimal
-from typing import Any, List, Sequence
+from typing import Any, List, Sequence, Union
+
+from ._logger import logger
 
 
 class RecordConvertor:
     @staticmethod
-    def __to_sqlite_element(value: Any) -> Any:
+    def __to_sqlite_element(value: Any, attr: Union[int, str]) -> Any:
         if isinstance(value, Decimal):
             return float(value)
 
+        if isinstance(value, int):
+            # INTEGER. The value is a signed integer,
+            # stored in 1, 2, 3, 4, 6, or 8 bytes depending on the magnitude of the value.
+            # https://www.sqlite.org/datatype3.html
+            if not (-9223372036854775808 < value < 9223372036854775807):
+                raise OverflowError(attr)
+
         return value
 
     @classmethod
@@ -34,12 +43,15 @@ def to_record(cls, attr_names: Sequence[str], values) -> List:
 
         try:
             # from a dictionary to a list
-            return [cls.__to_sqlite_element(values.get(attr_name)) for attr_name in attr_names]
+            return [
+                cls.__to_sqlite_element(values.get(attr_name), attr_name)
+                for attr_name in attr_names
+            ]
         except AttributeError:
             pass
 
         if isinstance(values, (tuple, list)):
-            return [cls.__to_sqlite_element(value) for value in values]
+            return [cls.__to_sqlite_element(value, col) for col, value in enumerate(values)]
 
         raise ValueError("cannot convert from {} to list".format(type(values)))
 
@@ -56,4 +68,26 @@ def to_records(cls, attr_names: Sequence[str], value_matrix: Sequence) -> List:
         .. seealso:: :py:meth:`.to_record`
         """
 
-        return [cls.to_record(attr_names, record) for record in value_matrix]
+        records = []
+        error_msgs = []
+
+        for row_idx, record in enumerate(value_matrix):
+            try:
+                records.append(cls.to_record(attr_names, record))
+            except OverflowError as e:
+                try:
+                    if isinstance(e.args[0], int):
+                        col_idx = e.args[0]
+                        col = "{} ({})".format(attr_names[col_idx], col_idx)
+                    else:
+                        col = e.args[0]
+                except IndexError as e:
+                    logger.error(e)
+                    continue
+
+                error_msgs.append("  overflow int found: row={}, col={}".format(row_idx, col))
+
+        if error_msgs:
+            raise OverflowError("failed to convert:\n" + "\n".join(error_msgs))
+
+        return records

test/test_convertor.py

@@ -77,3 +77,14 @@ def test_normal(self, attr_names, value, expected):
     def test_exception(self, attr_names, value, expected):
         with pytest.raises(expected):
             RecordConvertor.to_records(attr_names, value)
+
+    @pytest.mark.parametrize(
+        ["attr_names", "value"],
+        [
+            [["a", "b"], [[5, 9223372036854775808]]],
+            [["a", "b"], [{"a": -9223372036854775809, "b": 0}]],
+        ],
+    )
+    def test_exception_value(self, attr_names, value):
+        with pytest.raises(OverflowError):
+            RecordConvertor.to_records(attr_names, value)