Merge pull request #1064 from cmu-delphi/krivard/covid_hosp_datechecks

Fix problem with covid_hosp skipping state revisions.

Author: krivard

src/acquisition/covid_hosp/common/database.py

@@ -19,6 +19,7 @@ class Database:
   def __init__(self,
                connection,
                table_name=None,
+               hhs_dataset_id=None,
                columns_and_types=None,
                key_columns=None,
                additional_fields=None):
@@ -30,6 +31,8 @@ def __init__(self,
       An open connection to a database.
     table_name : str
       The name of the table which holds the dataset.
+    hhs_dataset_id : str
+      The 9-character healthdata.gov identifier for this dataset.
     columns_and_types : tuple[str, str, Callable]
       List of 3-tuples of (CSV header name, SQL column name, data type) for
       all the columns in the CSV file.
@@ -40,6 +43,7 @@ def __init__(self,
 
     self.connection = connection
     self.table_name = table_name
+    self.hhs_dataset_id = hhs_dataset_id
     self.publication_col_name = "issue" if table_name == 'covid_hosp_state_timeseries' else \
       'publication_date'
     self.columns_and_types = {
@@ -115,8 +119,8 @@ def contains_revision(self, revision):
         FROM
           `covid_hosp_meta`
         WHERE
-          `dataset_name` = %s AND `revision_timestamp` = %s
-      ''', (self.table_name, revision))
+          `hhs_dataset_id` = %s AND `revision_timestamp` = %s
+      ''', (self.hhs_dataset_id, revision))
       for (result,) in cursor:
         return bool(result)
 
@@ -138,14 +142,15 @@ def insert_metadata(self, publication_date, revision, meta_json):
         INSERT INTO
           `covid_hosp_meta` (
             `dataset_name`,
+            `hhs_dataset_id`,
             `publication_date`,
             `revision_timestamp`,
             `metadata_json`,
             `acquisition_datetime`
           )
         VALUES
-          (%s, %s, %s, %s, NOW())
-      ''', (self.table_name, publication_date, revision, meta_json))
+          (%s, %s, %s, %s, %s, NOW())
+      ''', (self.table_name, self.hhs_dataset_id, publication_date, revision, meta_json))
 
   def insert_dataset(self, publication_date, dataframe):
     """Add a dataset to the database.
@@ -232,7 +237,7 @@ def get_max_issue(self):
         from
           `covid_hosp_meta`
         WHERE
-          dataset_name = "{self.table_name}"
+          hhs_dataset_id = "{self.hhs_dataset_id}"
       ''')
       for (result,) in cursor:
         if result is not None:

src/acquisition/covid_hosp/common/utils.py

@@ -169,19 +169,20 @@ def update_dataset(database, network, newer_than=None, older_than=None):
       # download the dataset and add it to the database
       dataset = Utils.merge_by_key_cols([network.fetch_dataset(url) for url, _ in revisions],
                                         db.KEY_COLS)
-      # add metadata to the database using the last revision seen.
-      last_url, last_index = revisions[-1]
-      metadata_json = metadata.loc[last_index].reset_index().to_json()
+      # add metadata to the database
+      all_metadata = []
+      for url, index in revisions:
+        all_metadata.append((url, metadata.loc[index].reset_index().to_json()))
       datasets.append((
         issue_int,
         dataset,
-        last_url,
-        metadata_json
+        all_metadata
       ))
     with database.connect() as db:
-      for issue_int, dataset, last_url, metadata_json in datasets:
+      for issue_int, dataset, all_metadata in datasets:
         db.insert_dataset(issue_int, dataset)
-        db.insert_metadata(issue_int, last_url, metadata_json)
+        for url, metadata_json in all_metadata:
+          db.insert_metadata(issue_int, url, metadata_json)
         print(f'successfully acquired {len(dataset)} rows')
 
       # note that the transaction is committed by exiting the `with` block

src/acquisition/covid_hosp/facility/database.py

@@ -2,6 +2,7 @@
 from delphi.epidata.acquisition.covid_hosp.common.database import Database as BaseDatabase
 from delphi.epidata.acquisition.covid_hosp.common.database import Columndef
 from delphi.epidata.acquisition.covid_hosp.common.utils import Utils
+from delphi.epidata.acquisition.covid_hosp.facility.network import Network
 
 
 class Database(BaseDatabase):
@@ -213,5 +214,6 @@ def __init__(self, *args, **kwargs):
         *args,
         **kwargs,
         table_name=Database.TABLE_NAME,
+        hhs_dataset_id=Network.DATASET_ID,
         key_columns=Database.KEY_COLS,
         columns_and_types=Database.ORDERED_CSV_COLUMNS)

src/acquisition/covid_hosp/state_daily/database.py

@@ -2,6 +2,7 @@
 from delphi.epidata.acquisition.covid_hosp.common.database import Database as BaseDatabase
 from delphi.epidata.acquisition.covid_hosp.common.database import Columndef
 from delphi.epidata.acquisition.covid_hosp.common.utils import Utils
+from delphi.epidata.acquisition.covid_hosp.state_daily.network import Network
 
 
 class Database(BaseDatabase):
@@ -223,6 +224,7 @@ def __init__(self, *args, **kwargs):
         *args,
         **kwargs,
         table_name=Database.TABLE_NAME,
+        hhs_dataset_id=Network.DATASET_ID,
         columns_and_types=Database.ORDERED_CSV_COLUMNS,
         key_columns=Database.KEY_COLS,
         additional_fields=[Columndef('D', 'record_type', None)])

src/acquisition/covid_hosp/state_timeseries/database.py

@@ -2,6 +2,7 @@
 from delphi.epidata.acquisition.covid_hosp.common.database import Database as BaseDatabase
 from delphi.epidata.acquisition.covid_hosp.common.database import Columndef
 from delphi.epidata.acquisition.covid_hosp.common.utils import Utils
+from delphi.epidata.acquisition.covid_hosp.state_timeseries.network import Network
 
 
 class Database(BaseDatabase):
@@ -222,6 +223,7 @@ def __init__(self, *args, **kwargs):
         *args,
         **kwargs,
         table_name=Database.TABLE_NAME,
+        hhs_dataset_id=Network.DATASET_ID,
         columns_and_types=Database.ORDERED_CSV_COLUMNS,
         key_columns=Database.KEY_COLS,
         additional_fields=[Columndef('T', 'record_type', None)])

src/ddl/covid_hosp.sql

@@ -48,6 +48,7 @@ surfaced through the Epidata API.
 CREATE TABLE `covid_hosp_meta` (
   `id` INT NOT NULL AUTO_INCREMENT,
   `dataset_name` VARCHAR(64) NOT NULL,
+  `hhs_dataset_id` CHAR(9) NOT NULL DEFAULT "????-????",
   `publication_date` INT NOT NULL,
   `revision_timestamp` VARCHAR(512) NOT NULL,
   `metadata_json` JSON NOT NULL,

src/ddl/migrations/covid_hosp_meta_v0.4.4-v0.4.5.sql

@@ -0,0 +1,4 @@
+ALTER TABLE covid_hosp_meta ADD COLUMN hhs_dataset_id CHAR(9) NOT NULL DEFAULT "????-????";
+UPDATE covid_hosp_meta SET hhs_dataset_id="g62h-syeh" WHERE revision_timestamp LIKE "%g62h-syeh%";
+UPDATE covid_hosp_meta SET hhs_dataset_id="6xf2-c3ie" WHERE revision_timestamp LIKE "%6xf2-c3ie%";
+UPDATE covid_hosp_meta SET hhs_dataset_id="anag-cw7u" WHERE revision_timestamp LIKE "%anag-cw7u%";

tests/acquisition/covid_hosp/common/test_database.py

@@ -68,7 +68,7 @@ def test_contains_revision(self):
 
     mock_connection = MagicMock()
     mock_cursor = mock_connection.cursor()
-    database = Database(mock_connection, table_name=sentinel.table_name)
+    database = Database(mock_connection, table_name=sentinel.table_name, hhs_dataset_id=sentinel.hhs_dataset_id)
 
     with self.subTest(name='new revision'):
       mock_cursor.__iter__.return_value = [(0,)]
@@ -78,7 +78,7 @@ def test_contains_revision(self):
       # compare with boolean literal to test the type cast
       self.assertIs(result, False)
       query_values = mock_cursor.execute.call_args[0][-1]
-      self.assertEqual(query_values, (sentinel.table_name, sentinel.revision))
+      self.assertEqual(query_values, (sentinel.hhs_dataset_id, sentinel.revision))
 
     with self.subTest(name='old revision'):
       mock_cursor.__iter__.return_value = [(1,)]
@@ -88,7 +88,7 @@ def test_contains_revision(self):
       # compare with boolean literal to test the type cast
       self.assertIs(result, True)
       query_values = mock_cursor.execute.call_args[0][-1]
-      self.assertEqual(query_values, (sentinel.table_name, sentinel.revision))
+      self.assertEqual(query_values, (sentinel.hhs_dataset_id, sentinel.revision))
 
   def test_insert_metadata(self):
     """Add new metadata to the database."""
@@ -98,7 +98,7 @@ def test_insert_metadata(self):
 
     mock_connection = MagicMock()
     mock_cursor = mock_connection.cursor()
-    database = Database(mock_connection, table_name=sentinel.dataset_name)
+    database = Database(mock_connection, table_name=sentinel.table_name, hhs_dataset_id=sentinel.hhs_dataset_id)
 
     result = database.insert_metadata(
         sentinel.publication_date,
@@ -108,7 +108,8 @@ def test_insert_metadata(self):
     self.assertIsNone(result)
     actual_values = mock_cursor.execute.call_args[0][-1]
     expected_values = (
-      sentinel.dataset_name,
+      sentinel.table_name,
+      sentinel.hhs_dataset_id,
       sentinel.publication_date,
       sentinel.revision,
       sentinel.meta_json,

tests/acquisition/covid_hosp/common/test_utils.py

@@ -129,7 +129,10 @@ def test_run_acquire_new_dataset(self):
 
     self.assertTrue(result)
 
-    mock_connection.insert_metadata.assert_called_once()
+    # should have been called twice
+    mock_connection.insert_metadata.assert_called()
+    assert mock_connection.insert_metadata.call_count == 2
+    # most recent call should be for the final revision at url2
     args = mock_connection.insert_metadata.call_args[0]
     self.assertEqual(args[:2], (20210315, "url2"))
     pd.testing.assert_frame_equal(