From 23575874eb7b9fc9cca5cb07d87b05a748ad3b2d Mon Sep 17 00:00:00 2001
From: roll <eskarev@gmail.com>
Date: Thu, 23 Apr 2020 11:01:28 +0300
Subject: [PATCH] Added additional information to exceptions

---
 dataflows/base/datastream_processor.py | 13 +++++++++++--
 dataflows/base/flow.py                 |  4 ++--
 tests/test_lib.py                      | 14 ++++++++++++++
 3 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/dataflows/base/datastream_processor.py b/dataflows/base/datastream_processor.py
index ae71133..9854be7 100644
--- 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
index e2a8225..f17ad4d 100644
--- 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
index c6b119d..97e1ff0 100644
--- a/tests/test_lib.py
+++ b/tests/test_lib.py
@@ -1532,6 +1532,7 @@ def test_force_temporal_format():
         }
     ]]
 
+
 # Extract missing values
 
 def test_extract_missing_values():
@@ -1657,3 +1658,16 @@ def test_conditional():
     assert result2[0] == [
         dict(a=i, c=i) for i in range(3)
     ]
+
+
+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