Added additional information to exceptions

datahq · Apr 23, 2020 · cfb6765 · cfb6765
1 parent 9171565
commit cfb6765
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 4 deletions.
diff --git a/dataflows/base/datastream_processor.py b/dataflows/base/datastream_processor.py
@@ -26,11 +26,13 @@ def __init__(self):
         self.stats = {}
         self.source = None
         self.datapackage = None
+        self.position = None
 
-    def __call__(self, source=None):
+    def __call__(self, source=None, position=None):
         if source is None:
             source = DataStream()
         self.source = source
+        self.position = position
         return self
 
     def process_resource(self, resource: ResourceWrapper):
@@ -69,7 +71,14 @@ def func():
         return func
 
     def _process(self):
-        datastream = self.source._process()
+        try:
+            datastream = self.source._process()
+        except Exception as exception:
+            if (not hasattr(exception, 'processorName')):
+                exception.processorName = self.__class__.__name__
+                exception.processorObject = self
+                exception.processorPosition = self.position
+            raise exception
 
         self.datapackage = Package(descriptor=copy.deepcopy(datastream.dp.descriptor))
         self.datapackage = self.process_datapackage(self.datapackage)

diff --git a/dataflows/base/flow.py b/dataflows/base/flow.py
@@ -29,11 +29,11 @@ def _preprocess_chain(self):
     def _chain(self, ds=None):
         from ..helpers import datapackage_processor, rows_processor, row_processor, iterable_loader
 
-        for link in self._preprocess_chain():
+        for position, link in enumerate(self._preprocess_chain(), start=1):
             if isinstance(link, Flow):
                 ds = link._chain(ds)
             elif isinstance(link, DataStreamProcessor):
-                ds = link(ds)
+                ds = link(ds, position=position)
             elif isfunction(link):
                 sig = signature(link)
                 params = list(sig.parameters)

diff --git a/tests/test_lib.py b/tests/test_lib.py
@@ -1531,3 +1531,16 @@ def test_force_temporal_format():
             'time': datetime.time(8, 10, 4),
         }
     ]]
+
+
+def test_exception_information():
+    from dataflows import load
+    flow = Flow(
+        load('data/bad-path1.csv'),
+        load('data/bad-path2.csv'),
+    )
+    with pytest.raises(Exception) as excinfo:
+        data = flow.results()
+    assert excinfo.value.processorName == 'load'
+    assert excinfo.value.processorObject.load_source == 'data/bad-path2.csv'
+    assert excinfo.value.processorPosition == 2