v0.0.26 Only cast row values that are being tested (#36)

* Only cast row values that are being tested

* v0.0.26

* lint

dataflows/VERSION

@@ -1 +1 @@
-0.0.25
+0.0.26

dataflows/base/schema_validator.py

@@ -1,5 +1,3 @@
-import logging
-
 from datapackage import Resource
 from tableschema import Schema
 from tableschema.exceptions import CastError
@@ -19,21 +17,17 @@ def __init__(self, resource_name, row, index, cast_error):
         self.cast_error = cast_error
 
 
-def schema_validator(resource: Resource, iterator):
+def schema_validator(resource: Resource, iterator, field_names=None):
     schema: Schema = resource.schema
-    field_names = [f.name for f in schema.fields]
-    warned_fields = set()
+    if field_names is None:
+        field_names = [f.name for f in schema.fields]
+    schema_fields = [f for f in schema.fields if f.name in field_names]
     for i, row in enumerate(iterator):
-        to_cast = [row.get(f) for f in field_names]
+
         try:
-            casted = schema.cast_row(to_cast)
-            row = dict(zip(field_names, casted))
+            for f in schema_fields:
+                row[f.name] = f.cast_value(row[f.name])
         except CastError as e:
             raise ValidationError(resource.name, row, i, e)
 
-        for k in set(row.keys()) - set(field_names):
-            if k not in warned_fields:
-                warned_fields.add(k)
-                logging.warning('Encountered field %r, not in schema', k)
-
         yield row

dataflows/processors/set_type.py

@@ -11,11 +11,15 @@ def __init__(self, name, resources=-1, **options):
         self.name = re.compile(f'^{name}$')
         self.options = options
         self.resources = resources
+        self.field_names = []
 
     def process_resources(self, resources):
         for res in resources:
             if self.matcher.match(res.res.name):
-                yield schema_validator(res.res, res)
+                if len(self.field_names) > 0:
+                    yield schema_validator(res.res, res, field_names=self.field_names)
+                else:
+                    yield res
             else:
                 yield res
 
@@ -28,6 +32,7 @@ def process_datapackage(self, dp):
                 for field in res['schema']['fields']:
                     if self.name.match(field['name']):
                         field.update(self.options)
+                        self.field_names.append(field['name'])
                         added = True
         assert added, 'Failed to find field {} in schema'.format(self.name)
         return dp

tests/test_examples.py

@@ -139,7 +139,7 @@ def test_example_7():
     def add_is_guitarist_column(package):
 
         # Add a new field to the first resource
-        package.pkg.resources[0].descriptor['schema']['fields'].append(dict(
+        package.pkg.descriptor['resources'][0]['schema']['fields'].append(dict(
             name='is_guitarist',
             type='boolean'
         ))
@@ -171,7 +171,7 @@ def test_example_75():
     def add_is_guitarist_column_to_schema(package):
 
         # Add a new field to the first resource
-        package.pkg.resources[0].descriptor['schema']['fields'].append(dict(
+        package.pkg.descriptor['resources'][0]['schema']['fields'].append(dict(
             name='is_guitarist',
             type='boolean'
         ))