Repository: frictionlessdata/frictionless-py

Issue:
Introduce `on_error` parameter for row reading
# Overview

At the moment there are 3 types of streams in frictionless:
- byte stream (nothing to validate as it's the source)
- data stream (nothing to validate as it's the source)
- row stream (validated against a schema)

Current implementation cleans rows using a schema (replacing errored cells by nulls) and adds errors to `row.errors` and the user is responsible on reacting on row errors. This behaviour is used by many csv parsers like `papaparse.js` and others:

```python
resource = Resource(path='data/table.csv')
for row in resource.read_rows():
  print(row.errors)
  print(row)
```

But for some scenarios when we can't validate data in advance using the `validate` function we want to guarantee data validness doing some data reading e.g. export:

```python
resource = Resource(path='data/table.csv')
resource.to_sql(...) # I need a guarantee that data is 100% valid
```

So we need a flag like `on_error` added to reading functions or constructors e.g.:

```python
resource = Resource(path='data/table.csv', on_error='ignore')
resource = Resource(path='data/table.csv', on_error='warn')
resource = Resource(path='data/table.csv', on_error='raise')
resource = Resource(path='data/table.csv', on_error=function)
```

Also, we need to think about what behavior needs to be a default. It can be stricter `raise` aka fail fast as it's used in current `tableschema/datapackage` or more forgiving `ignore`, the current frictionless behavior. Probably stricter can be `less surprising` behavior for most of the users.


Discussion:

Patch:
diff --git a/frictionless/extract/table.py b/frictionless/extract/table.py
index d258f81e..19f4ad22 100644
--- a/frictionless/extract/table.py
+++ b/frictionless/extract/table.py
@@ -12,11 +12,12 @@ def extract_table(
     encoding=None,
     compression=None,
     compression_path=None,
+    # Control/Dialect/Query/Header
     control=None,
-    # Table
     dialect=None,
     query=None,
     headers=None,
+    # Schema
     schema=None,
     sync_schema=False,
     patch_schema=False,
@@ -25,6 +26,8 @@ def extract_table(
     infer_volume=config.DEFAULT_INFER_VOLUME,
     infer_confidence=config.DEFAULT_INFER_CONFIDENCE,
     infer_missing_values=config.DEFAULT_MISSING_VALUES,
+    # Integrity
+    on_error="ignore",
     lookup=None,
     # Extraction
     process=None,
@@ -112,8 +115,11 @@ def extract_table(
         lookup? (dict): The lookup is a special object providing relational information.
             For more information, please check "Extracting  Data" guide.
 
-        process? (func): a row processor function
-        stream? (bool): return a row streams instead of loading into memory
+        process? (func): A row processor function.
+            It should be in a form of `row -> result`
+
+        stream? (bool): Return a row streams instead of loading into memory.
+            It can be useful for a big data files.
 
     Returns:
         Row[]: an array/stream of rows
@@ -123,7 +129,6 @@ def extract_table(
     # Create table
     table = Table(
         source,
-        headers=headers,
         # File
         scheme=scheme,
         format=format,
@@ -131,9 +136,11 @@ def extract_table(
         encoding=encoding,
         compression=compression,
         compression_path=compression_path,
+        # Control/Dialect/Query/Header
         control=control,
         dialect=dialect,
         query=query,
+        headers=headers,
         # Schema
         schema=schema,
         sync_schema=sync_schema,
@@ -142,6 +149,8 @@ def extract_table(
         infer_names=infer_names,
         infer_volume=infer_volume,
         infer_confidence=infer_confidence,
+        # Integrity
+        on_error=on_error,
         lookup=lookup,
     )
 
diff --git a/frictionless/package.py b/frictionless/package.py
index a7919bf6..59a69513 100644
--- a/frictionless/package.py
+++ b/frictionless/package.py
@@ -36,6 +36,7 @@ class Package(Metadata):
         profile? (str): profile name like 'data-package'
         basepath? (str): a basepath of the package
         trusted? (bool): don't raise on unsafe paths
+        on_error? (ignore|warn|raise): behaviour if there is an error
 
     Raises:
         FrictionlessException: raise any error that occurs during the process
@@ -53,6 +54,7 @@ class Package(Metadata):
         profile=None,
         basepath=None,
         trusted=None,
+        on_error="ignore",
     ):
 
         # Handle zip
@@ -67,8 +69,15 @@ class Package(Metadata):
         self.setinitial("profile", profile)
         self.__basepath = basepath or helpers.detect_basepath(descriptor)
         self.__trusted = trusted
+        self.__on_error = on_error
         super().__init__(descriptor)
 
+    def __setattr__(self, name, value):
+        if name == "on_error":
+            self.__on_error = value
+            return
+        super().__setattr__(name, value)
+
     @Metadata.property
     def name(self):
         """
@@ -109,6 +118,15 @@ class Package(Metadata):
         """
         return self.get("profile", config.DEFAULT_PACKAGE_PROFILE)
 
+    @property
+    def on_error(self):
+        """
+        Returns:
+            ignore|warn|raise: on error bahaviour
+        """
+        assert self.__on_error in ["ignore", "warn", "raise"]
+        return self.__on_error
+
     # Resources
 
     @Metadata.property
@@ -422,9 +440,12 @@ class Package(Metadata):
                         resource,
                         basepath=self.__basepath,
                         trusted=self.__trusted,
+                        on_error=self.__on_error,
                         package=self,
                     )
                     list.__setitem__(resources, index, resource)
+                # NOTE: should we sync basepath/trusted also here?
+                resource.on_error = self.__on_error
             if not isinstance(resources, helpers.ControlledList):
                 resources = helpers.ControlledList(resources)
                 resources.__onchange__(self.metadata_process)
diff --git a/frictionless/resource.py b/frictionless/resource.py
index 6a5f43fe..aa4b3e4c 100644
--- a/frictionless/resource.py
+++ b/frictionless/resource.py
@@ -42,6 +42,7 @@ class Resource(Metadata):
         profile? (str): resource profile
         basepath? (str): resource basepath
         trusted? (bool): don't raise on unsage paths
+        on_error? (ignore|warn|raise): behaviour if there is an error
         package? (Package): resource package
 
     Raises:
@@ -69,6 +70,7 @@ class Resource(Metadata):
         profile=None,
         basepath=None,
         trusted=False,
+        on_error="ignore",
         package=None,
     ):
 
@@ -93,6 +95,7 @@ class Resource(Metadata):
         self.setinitial("profile", profile)
         self.__basepath = basepath or helpers.detect_basepath(descriptor)
         self.__trusted = trusted
+        self.__on_error = on_error
         self.__package = package
         super().__init__(descriptor)
 
@@ -101,6 +104,12 @@ class Resource(Metadata):
         if hashing != config.DEFAULT_HASHING:
             self["hashing"] = hashing
 
+    def __setattr__(self, name, value):
+        if name == "on_error":
+            self.__on_error = value
+            return
+        super().__setattr__(name, value)
+
     @Metadata.property
     def name(self):
         """
@@ -286,21 +295,6 @@ class Resource(Metadata):
         """
         return self.get("compressionPath")
 
-    @Metadata.property
-    def stats(self):
-        """
-        Returns
-            dict?: resource stats
-        """
-        stats = {}
-        for name in ["hash", "bytes", "rows"]:
-            value = self.get(name)
-            if value is not None:
-                if name == "hash":
-                    value = helpers.parse_resource_hash(value)[1]
-                stats[name] = value
-        return stats
-
     @Metadata.property
     def dialect(self):
         """
@@ -341,6 +335,30 @@ class Resource(Metadata):
         """
         return self.get("profile", config.DEFAULT_RESOURCE_PROFILE)
 
+    @property
+    def on_error(self):
+        """
+        Returns:
+            ignore|warn|raise: on error bahaviour
+        """
+        assert self.__on_error in ["ignore", "warn", "raise"]
+        return self.__on_error
+
+    @Metadata.property
+    def stats(self):
+        """
+        Returns
+            dict?: resource stats
+        """
+        stats = {}
+        for name in ["hash", "bytes", "rows"]:
+            value = self.get(name)
+            if value is not None:
+                if name == "hash":
+                    value = helpers.parse_resource_hash(value)[1]
+                stats[name] = value
+        return stats
+
     # Expand
 
     def expand(self):
@@ -724,6 +742,7 @@ class Resource(Metadata):
         options.setdefault("compression_path", self.compression_path)
         options.setdefault("dialect", self.dialect)
         options.setdefault("schema", self.schema)
+        options.setdefault("on_error", self.__on_error)
         if "lookup" not in options:
             options["lookup"] = self.read_lookup()
         return Table(**options)
diff --git a/frictionless/table.py b/frictionless/table.py
index e02471b3..b4fcc307 100644
--- a/frictionless/table.py
+++ b/frictionless/table.py
@@ -1,4 +1,5 @@
 import typing
+import warnings
 from pathlib import Path
 from copy import deepcopy
 from itertools import chain
@@ -105,6 +106,10 @@ class Table:
             For more information, please check "Describing  Data" guide.
             It defaults to `['']`
 
+        on_error? (ignore|warn|raise): Define behaviour if there is an error in the
+            header or rows during the reading rows process.
+            It defaults to `ignore`.
+
         lookup? (dict): The lookup is a special object providing relational information.
             For more information, please check "Extracting  Data" guide.
 
@@ -123,11 +128,12 @@ class Table:
         encoding=None,
         compression=None,
         compression_path=None,
+        # Control/Dialect/Query/Header
         control=None,
-        # Table
         dialect=None,
         query=None,
         headers=None,
+        # Schema
         schema=None,
         sync_schema=False,
         patch_schema=False,
@@ -136,6 +142,8 @@ class Table:
         infer_volume=config.DEFAULT_INFER_VOLUME,
         infer_confidence=config.DEFAULT_INFER_CONFIDENCE,
         infer_missing_values=config.DEFAULT_MISSING_VALUES,
+        # Integrity
+        on_error="ignore",
         lookup=None,
     ):
 
@@ -177,8 +185,12 @@ class Table:
         self.__infer_volume = infer_volume
         self.__infer_confidence = infer_confidence
         self.__infer_missing_values = infer_missing_values
+        self.__on_error = on_error
         self.__lookup = lookup
 
+        # Set error handler
+        self.on_error = on_error
+
         # Create file
         self.__file = File(
             source=source,
@@ -193,6 +205,12 @@ class Table:
             query=query,
         )
 
+    def __setattr__(self, name, value):
+        if name == "on_error":
+            self.__on_error = value
+            return
+        super().__setattr__(name, value)
+
     def __enter__(self):
         if self.closed:
             self.open()
@@ -301,6 +319,24 @@ class Table:
         """
         return self.__schema
 
+    @property
+    def on_error(self):
+        """
+        Returns:
+            ignore|warn|raise: on error bahaviour
+        """
+        assert self.__on_error in ["ignore", "warn", "raise"]
+        return self.__on_error
+
+    # TODO: use property wrapper to make it shorter
+    @on_error.setter
+    def on_error(self, value):
+        """
+        Parameters:
+            value (ignore|warn|raise): on error bahaviour
+        """
+        self.__on_error = value
+
     @property
     def header(self):
         """
@@ -403,7 +439,8 @@ class Table:
         """
         return self.__parser is None
 
-    # Read
+        # Read
+        return self.__on_error
 
     def read_data(self):
         """Read data stream into memory
@@ -662,6 +699,14 @@ class Table:
     def __read_row_stream_create(self):
         schema = self.schema
 
+        # Handle header errors
+        if not self.header.valid:
+            error = self.header.errors[0]
+            if self.__on_error == "warn":
+                warnings.warn(error.message, UserWarning)
+            elif self.__on_error == "raise":
+                raise exceptions.FrictionlessException(error)
+
         # Create state
         memory_unique = {}
         memory_primary = {}
@@ -732,6 +777,14 @@ class Table:
                             error = errors.ForeignKeyError.from_row(row, note=note)
                             row.errors.append(error)
 
+            # Handle row errors
+            if not row.valid:
+                error = row.errors[0]
+                if self.__on_error == "warn":
+                    warnings.warn(error.message, UserWarning)
+                elif self.__on_error == "raise":
+                    raise exceptions.FrictionlessException(error)
+
             # Stream row
             yield row
 


Test patch:
diff --git a/tests/test_package.py b/tests/test_package.py
index 2a2e58f2..c2fea3ed 100644
--- a/tests/test_package.py
+++ b/tests/test_package.py
@@ -2,7 +2,7 @@ import os
 import json
 import zipfile
 import pytest
-from frictionless import Package, exceptions
+from frictionless import Package, Resource, exceptions
 
 
 # General
@@ -500,7 +500,60 @@ def test_package_compression_explicit_zip():
 
 # Integrity
 
-INTEGRITY_DESCRIPTOR = {
+
+def test_resource_integrity_on_error():
+    package = Package(resources=[Resource(path="data/invalid.csv")])
+    resource = package.resources[0]
+    assert package.on_error == "ignore"
+    assert resource.on_error == "ignore"
+    assert resource.read_rows()
+
+
+def test_resource_integrity_on_error_header_warn():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"name": "bad"}]}
+    package = Package(resources=[Resource(data=data, schema=schema)], on_error="warn")
+    resource = package.resources[0]
+    assert package.on_error == "warn"
+    assert resource.on_error == "warn"
+    with pytest.warns(UserWarning):
+        resource.read_rows()
+
+
+def test_resource_integrity_on_error_header_raise():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"name": "bad"}]}
+    package = Package({"resources": [{"data": data, "schema": schema}]}, on_error="raise")
+    resource = package.resources[0]
+    assert package.on_error == "raise"
+    assert resource.on_error == "raise"
+    with pytest.raises(exceptions.FrictionlessException):
+        resource.read_rows()
+
+
+def test_resource_integrity_on_error_row_warn():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"type": "string"}]}
+    package = Package(resources=[Resource(data=data, schema=schema)], on_error="warn")
+    resource = package.resources[0]
+    assert package.on_error == "warn"
+    assert resource.on_error == "warn"
+    with pytest.warns(UserWarning):
+        resource.read_rows()
+
+
+def test_resource_integrity_on_error_row_raise():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"type": "string"}]}
+    package = Package({"resources": [{"data": data, "schema": schema}]}, on_error="raise")
+    resource = package.resources[0]
+    assert package.on_error == "raise"
+    assert resource.on_error == "raise"
+    with pytest.raises(exceptions.FrictionlessException):
+        resource.read_rows()
+
+
+DESCRIPTOR_FK = {
     "resources": [
         {
             "name": "main",
@@ -538,8 +591,8 @@ INTEGRITY_DESCRIPTOR = {
 }
 
 
-def test_package_integrity():
-    package = Package(INTEGRITY_DESCRIPTOR)
+def test_package_integrity_foreign_key():
+    package = Package(DESCRIPTOR_FK)
     resource = package.get_resource("main")
     rows = resource.read_rows()
     assert rows[0].valid
@@ -553,7 +606,7 @@ def test_package_integrity():
 
 
 def test_package_integrity_foreign_key_invalid():
-    package = Package(INTEGRITY_DESCRIPTOR)
+    package = Package(DESCRIPTOR_FK)
     package.resources[1].data[3][0] = "bad"
     resource = package.get_resource("main")
     rows = resource.read_rows()
@@ -568,7 +621,7 @@ def test_package_integrity_foreign_key_invalid():
 
 
 def test_package_integrity_foreign_key_self_reference():
-    package = Package(INTEGRITY_DESCRIPTOR)
+    package = Package(DESCRIPTOR_FK)
     package.resources[0].schema.foreign_keys = [
         {"fields": "parent_id", "reference": {"resource": "", "fields": "id"}}
     ]
@@ -580,7 +633,7 @@ def test_package_integrity_foreign_key_self_reference():
 
 
 def test_package_integrity_foreign_key_self_reference_invalid():
-    package = Package(INTEGRITY_DESCRIPTOR)
+    package = Package(DESCRIPTOR_FK)
     package.resources[0].data[2][0] = "0"
     package.resources[0].schema.foreign_keys = [
         {"fields": "parent_id", "reference": {"resource": "", "fields": "id"}}
@@ -593,7 +646,7 @@ def test_package_integrity_foreign_key_self_reference_invalid():
 
 
 def test_package_integrity_foreign_key_multifield():
-    package = Package(INTEGRITY_DESCRIPTOR)
+    package = Package(DESCRIPTOR_FK)
     package.resources[0].schema.foreign_keys = [
         {
             "fields": ["name", "surname"],
@@ -608,7 +661,7 @@ def test_package_integrity_foreign_key_multifield():
 
 
 def test_package_integrity_foreign_key_multifield_invalid():
-    package = Package(INTEGRITY_DESCRIPTOR)
+    package = Package(DESCRIPTOR_FK)
     package.resources[0].schema.foreign_keys = [
         {
             "fields": ["name", "surname"],
@@ -624,7 +677,7 @@ def test_package_integrity_foreign_key_multifield_invalid():
 
 
 def test_package_integrity_read_lookup():
-    package = Package(INTEGRITY_DESCRIPTOR)
+    package = Package(DESCRIPTOR_FK)
     resource = package.get_resource("main")
     lookup = resource.read_lookup()
     assert lookup == {"people": {("firstname",): {("Walter",), ("Alex",), ("John",)}}}
diff --git a/tests/test_resource.py b/tests/test_resource.py
index c62d7e14..bd4c6e96 100644
--- a/tests/test_resource.py
+++ b/tests/test_resource.py
@@ -794,7 +794,49 @@ def test_resource_source_multipart_infer():
 # Integrity
 
 
-INTEGRITY_DESCRIPTOR = {
+def test_resource_integrity_on_error():
+    resource = Resource(path="data/invalid.csv")
+    assert resource.on_error == "ignore"
+    assert resource.read_rows()
+
+
+def test_resource_integrity_on_error_header_warn():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"name": "bad"}]}
+    resource = Resource(data=data, schema=schema, on_error="warn")
+    assert resource.on_error == "warn"
+    with pytest.warns(UserWarning):
+        resource.read_rows()
+
+
+def test_resource_integrity_on_error_header_raise():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"name": "bad"}]}
+    resource = Resource(data=data, schema=schema, on_error="raise")
+    assert resource.on_error == "raise"
+    with pytest.raises(exceptions.FrictionlessException):
+        resource.read_rows()
+
+
+def test_resource_integrity_on_error_row_warn():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"type": "string"}]}
+    resource = Resource(data=data, schema=schema, on_error="warn")
+    assert resource.on_error == "warn"
+    with pytest.warns(UserWarning):
+        resource.read_rows()
+
+
+def test_resource_integrity_on_error_row_raise():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"type": "string"}]}
+    resource = Resource(data=data, schema=schema, on_error="raise")
+    assert resource.on_error == "raise"
+    with pytest.raises(exceptions.FrictionlessException):
+        resource.read_rows()
+
+
+DESCRIPTOR_FK = {
     "path": "data/nested.csv",
     "schema": {
         "fields": [
@@ -807,8 +849,8 @@ INTEGRITY_DESCRIPTOR = {
 }
 
 
-def test_resource_integrity():
-    resource = Resource(INTEGRITY_DESCRIPTOR)
+def test_resource_integrity_foreign_keys():
+    resource = Resource(DESCRIPTOR_FK)
     rows = resource.read_rows()
     assert rows[0].valid
     assert rows[1].valid
@@ -822,8 +864,8 @@ def test_resource_integrity():
     ]
 
 
-def test_resource_integrity_invalid():
-    resource = Resource(INTEGRITY_DESCRIPTOR, path="data/nested-invalid.csv")
+def test_resource_integrity_foreign_keys_invalid():
+    resource = Resource(DESCRIPTOR_FK, path="data/nested-invalid.csv")
     rows = resource.read_rows()
     assert rows[0].valid
     assert rows[1].valid
@@ -840,7 +882,7 @@ def test_resource_integrity_invalid():
 
 
 def test_resource_integrity_read_lookup():
-    resource = Resource(INTEGRITY_DESCRIPTOR)
+    resource = Resource(DESCRIPTOR_FK)
     lookup = resource.read_lookup()
     assert lookup == {"": {("id",): {(1,), (2,), (3,), (4,)}}}
 
diff --git a/tests/test_table.py b/tests/test_table.py
index 10d9e96c..885e1aa4 100644
--- a/tests/test_table.py
+++ b/tests/test_table.py
@@ -1246,6 +1246,48 @@ def test_table_write_format_error_bad_format(tmpdir):
 # Integrity
 
 
+def test_table_integrity_on_error():
+    with Table("data/invalid.csv") as table:
+        assert table.on_error == "ignore"
+        assert table.read_rows()
+
+
+def test_table_integrity_on_error_header_warn():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"name": "bad"}]}
+    with Table(data, schema=schema, on_error="warn") as table:
+        assert table.on_error == "warn"
+        with pytest.warns(UserWarning):
+            table.read_rows()
+
+
+def test_table_integrity_on_error_header_raise():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"name": "bad"}]}
+    with Table(data, schema=schema, on_error="raise") as table:
+        assert table.on_error == "raise"
+        with pytest.raises(exceptions.FrictionlessException):
+            table.read_rows()
+
+
+def test_table_integrity_on_error_row_warn():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"type": "string"}]}
+    with Table(data, schema=schema, on_error="warn") as table:
+        assert table.on_error == "warn"
+        with pytest.warns(UserWarning):
+            table.read_rows()
+
+
+def test_table_integrity_on_error_row_raise():
+    data = [["name"], [1], [2], [3]]
+    schema = {"fields": [{"type": "string"}]}
+    with Table(data, schema=schema, on_error="raise") as table:
+        assert table.on_error == "raise"
+        with pytest.raises(exceptions.FrictionlessException):
+            table.read_rows()
+
+
 def test_table_integrity_unique():
     source = [["name"], [1], [2], [3]]
     patch_schema = {"fields": {"name": {"constraints": {"unique": True}}}}