[326 precursor] Factored in utils from 326_rename (#329)

* file renaming factored from 326_nih

* revert autobatch

* removed references to old health_data terminology in favour of nih

* traded lambda for partial

* factored out utils

* factored in utils from 326_rename

* added docstrings and comments, and small amount of refactoring

* added pk tests

* test for auto pkey check

* test for generate pk

* added retrieve pk tests

* added delete stmt test

* added delete stmt test

* added merge tests

* added bucket keys tests

* backwards compatability for older pipelines

* fixes wrt to new insert method

* added minor fix to gtr routine due to tanzania

* added sleeps to test to avoid overhitting the api

nesta/core/batchables/gtr/collect_gtr/run.py

@@ -19,6 +19,7 @@
 from nesta.packages.gtr.get_gtr_data import TOP_URL
 
 from nesta.core.orms.orm_utils import insert_data
+from nesta.core.orms.orm_utils import orm_column_names
 from nesta.core.orms.orm_utils import get_class_by_tablename
 from nesta.core.orms.gtr_orm import Base
 from nesta.core.luigihacks.s3 import parse_s3_path
@@ -54,9 +55,10 @@ def run():
     objs = []
     for table_name, rows in data.items():
         _class = get_class_by_tablename(Base, f"gtr_{table_name}")
-        # Remove any fields that aren't in the ORM
-        cleaned_rows = [{k:v for k, v in row.items() if k in _class.__dict__}
-                        for row in rows]
+        # Remove any fields that aren't in the ORM and set NULL as default
+        field_names = orm_column_names(_class)
+        cleaned_rows = [{field: (row[field] if field in row else None)
+                         for field in field_names} for row in rows]
         objs += insert_data("BATCHPAR_config", "mysqldb", db,
                             Base, _class, cleaned_rows)
 
@@ -73,9 +75,8 @@ def run():
 if __name__ == "__main__":
     # Local testing
     if "BATCHPAR_outinfo" not in os.environ:
-        os.environ['BATCHPAR_TOPURL'] = "https://gtr.ukri.org/gtr/api/projects"
-        os.environ['BATCHPAR_PAGESIZE'] = "100"
-        os.environ['BATCHPAR_page'] = "647"
+        os.environ['BATCHPAR_PAGESIZE'] = "10"
+        os.environ['BATCHPAR_page'] = "2"
         os.environ["BATCHPAR_db"] = "dev"
         os.environ["BATCHPAR_outinfo"] = ""
         os.environ["BATCHPAR_config"] = os.environ["MYSQLDBCONF"]

nesta/core/luigihacks/misctools.py

@@ -6,7 +6,7 @@
 from functools import lru_cache
 import yaml
 from datetime import datetime as dt
-
+import boto3
 
 def get_config(file_name, header, path="core/config/"):
     '''Get the configuration from a file in the luigi config path
@@ -125,3 +125,18 @@ def extract_task_info(luigi_task):
     task_name = type(luigi_task).__name__
     routine_id = f'{task_name}-{luigi_task.date}-{test}'
     return test, routine_id
+
+
+@lru_cache()
+def bucket_keys(bucket_name):
+    """Get all keys in an S3 bucket.
+    
+    Args:
+        bucket_name (str): Name of a bucket to query.
+    Returns:
+        keys (set): Set of keys
+    """
+    s3 = boto3.resource('s3')
+    bucket = s3.Bucket(bucket_name)
+    keys = set(obj.key for obj in bucket.objects.all())
+    return keys

nesta/core/luigihacks/tests/test_elasticsearchplus.py

@@ -2,6 +2,7 @@
 from unittest import mock
 from alphabet_detector import AlphabetDetector
 from collections import Counter
+import time
 
 from nesta.core.luigihacks.elasticsearchplus import Translator
 
@@ -94,6 +95,7 @@ def test_sentence_chunks():
 
 def test_auto_translate_true_short(row):
     """The translator shouldn't be applied for short pieces of text"""
+    time.sleep(5)
     _row = _auto_translate(row, TRANSLATOR, 1000)
     assert not _row.pop(TRANS_TAG)
     assert len(_row.pop(LANGS_TAG)) == 0
@@ -102,7 +104,9 @@ def test_auto_translate_true_short(row):
     assert row == _row
 
 def test_auto_translate_true_long_small_chunks(row):
+    time.sleep(5)
     _row_1 = _auto_translate(row, TRANSLATOR, 10, chunksize=1)
+    time.sleep(5)
     _row_2 = _auto_translate(row, TRANSLATOR, 10, chunksize=10000)
     assert _row_1.pop('mixed_lang') != _row_2.pop('mixed_lang')
 
@@ -129,6 +133,7 @@ def test_auto_translate_true_long_small_chunks(row):
     assert _row_1 == _row_2
 
 def test_auto_translate_true_long(row):
+    time.sleep(5)
     _row = _auto_translate(row, TRANSLATOR, 10)
     assert row.pop('korean') != _row['korean']
     assert row.pop('mixed_lang') != _row['mixed_lang']
@@ -147,6 +152,7 @@ def test_auto_translate_true_long(row):
 def test_auto_translate_false(row):
     row.pop('korean')
     row.pop('mixed_lang')
+    time.sleep(5)
     _row = _auto_translate(row, TRANSLATOR)
     assert not _row.pop(TRANS_TAG)
     assert _row.pop(LANGS_TAG) == ['en']

nesta/core/luigihacks/tests/test_misctools.py

@@ -1,6 +1,7 @@
-from unittest import TestCase
+from unittest import TestCase, mock
 from nesta.core.luigihacks.misctools import get_config
 from nesta.core.luigihacks.misctools import find_filepath_from_pathstub
+from nesta.core.luigihacks.misctools import bucket_keys
 
 
 class TestMiscTools(TestCase):
@@ -15,3 +16,20 @@ def test_find_filepath_from_pathstub(self):
         find_filepath_from_pathstub("nesta/packages")
         with self.assertRaises(FileNotFoundError):
             find_filepath_from_pathstub("nesta/package")
+
+
+@mock.patch('nesta.core.luigihacks.misctools.boto3')
+def test_bucket_keys(mocked_boto3):
+    keys = {'foo', 'bar', 'baz'}
+
+    # Mock up the bucket
+    bucket_objs = []
+    for key in keys:
+        obj = mock.Mock()
+        obj.key = key
+        bucket_objs.append(obj)
+    mocked_bucket = mocked_boto3.resource().Bucket()
+    mocked_bucket.objects.all.return_value = bucket_objs
+
+    # Actually do the test
+    assert bucket_keys('dummy') == keys