Create manage_revision fields as dates

Author: akariv

datapackage_pipelines_budgetkey/generator.py

@@ -146,6 +146,7 @@ def generate_pipeline(cls, source, base):
                             }
                         }
                     }),
+                    ('set-revisions', {}),
                     ('filter', {
                         'resources': doc_type,
                         'in': [

datapackage_pipelines_budgetkey/pipelines/budgetkey/elasticsearch/budgetkey.source-spec.yaml

@@ -112,7 +112,7 @@ supports:
 
 tenders:
   kind: indexer
-  revision: 15
+  revision: 16
   dependent_pipeline: ./procurement/tenders/processed
   source_datapackage: /var/datapackages/procurement/tenders/processed/datapackage.json
   page-title-pattern: '{snippet}'

datapackage_pipelines_budgetkey/processors/manage-revisions.py

@@ -87,6 +87,7 @@ def process_resource(res, key_fields, hash_fields, existing_ids, prefix):
             logging.info('#%d: KEY: %r HASH: %r', i, key, hash)
         try:
             existing_id = existing_ids.get(key)
+            row.update(existing_id)
             days_since_last_update = (now - existing_id[prefix+'__last_updated_at']).days
             days_since_last_update = max(0, days_since_last_update)
             next_update_days = existing_id[prefix+'__next_update_days']

datapackage_pipelines_budgetkey/processors/set-revision.py

@@ -18,9 +18,11 @@ def process_resources(res_iter_):
     yield from res_iter_
 
 
-dp['resources'][0]['schema']['fields'].append({
-    'name': '__revision',
-    'type': 'integer'
-})
+dp['resources'][0]['schema']['fields'].extend([
+    {
+        'name': '__revision',
+        'type': 'integer'
+    },
+])
 
 spew(dp, process_resources(res_iter))

datapackage_pipelines_budgetkey/processors/set-revisions.py

@@ -0,0 +1,27 @@
+from datapackage_pipelines.utilities.resources import PROP_STREAMING
+from datapackage_pipelines.wrapper import ingest, spew
+
+parameters, dp, res_iter = ingest()
+
+
+def process_resource(res_):
+    for row in res_:
+        for k in ['__last_updated_at', '__last_modified_at', '__created_at']:
+            if k in row and row[k]:
+                row['rev'+k[1:]] = row[k].date()
+        yield row
+
+
+def process_resources(res_iter_):
+    first = next(res_iter_)
+    yield process_resource(first)
+    yield from res_iter_
+
+
+dp['resources'][0]['schema']['fields'].extend([
+    {'name': 'rev_last_updated_at', 'type': 'date'},
+    {'name': 'rev_last_modified_at', 'type': 'date'},
+    {'name': 'rev_created_at', 'type': 'date'},
+])
+
+spew(dp, process_resources(res_iter))