Convert covid_hosp to use structured logger

* also adds some previously-absent logging

Author: krivard

src/acquisition/covid_hosp/common/database.py

@@ -124,7 +124,7 @@ def contains_revision(self, revision):
       for (result,) in cursor:
         return bool(result)
 
-  def insert_metadata(self, publication_date, revision, meta_json):
+  def insert_metadata(self, publication_date, revision, meta_json, logger=False):
     """Add revision metadata to the database.
 
     Parameters
@@ -135,6 +135,8 @@ def insert_metadata(self, publication_date, revision, meta_json):
       Unique revision string.
     meta_json : str
       Metadata serialized as a JSON string.
+    logger structlog.Logger [optional; default False]
+      Logger to receive messages
     """
 
     with self.new_cursor() as cursor:
@@ -152,7 +154,7 @@ def insert_metadata(self, 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):
+  def insert_dataset(self, publication_date, dataframe, logger=False):
     """Add a dataset to the database.
 
     Parameters
@@ -161,6 +163,8 @@ def insert_dataset(self, publication_date, dataframe):
       Date when the dataset was published in YYYYMMDD format.
     dataframe : pandas.DataFrame
       The dataset.
+    logger structlog.Logger [optional; default False]
+      Logger to receive messages.
     """
     dataframe_columns_and_types = [
       x for x in self.columns_and_types.values() if x.csv_name in dataframe.columns
@@ -181,6 +185,8 @@ def nan_safe_dtype(dtype, value):
     sql = f'INSERT INTO `{self.table_name}` (`id`, `{self.publication_col_name}`, {columns}) ' \
           f'VALUES ({value_placeholders})'
     id_and_publication_date = (0, publication_date)
+    if logger:
+      logger.info("updating values")
     with self.new_cursor() as cursor:
       for _, row in dataframe.iterrows():
         values = []
@@ -193,6 +199,8 @@ def nan_safe_dtype(dtype, value):
 
     # deal with non/seldomly updated columns used like a fk table (if this database needs it)
     if hasattr(self, 'AGGREGATE_KEY_COLS'):
+      if logger:
+        logger.info("updating keys")
       ak_cols = self.AGGREGATE_KEY_COLS
 
       # restrict data to just the key columns and remove duplicate rows
@@ -225,7 +233,7 @@ def nan_safe_dtype(dtype, value):
         cur.executemany(ak_insert_sql, ak_data)
 
 
-  def get_max_issue(self):
+  def get_max_issue(self, logger=False):
     """Fetch the most recent issue.
 
     This is used to bookend what updates we pull in from the HHS metadata.
@@ -242,4 +250,6 @@ def get_max_issue(self):
       for (result,) in cursor:
         if result is not None:
           return pd.Timestamp(str(result))
+      if logger:
+        logger.info("get_max_issue", msg="no matching results in meta table; returning 1900/1/1 epoch")
       return pd.Timestamp("1900/1/1")

src/acquisition/covid_hosp/common/network.py

@@ -6,7 +6,7 @@ class Network:
   METADATA_URL_TEMPLATE = \
       'https://healthdata.gov/api/views/%s/rows.csv'
 
-  def fetch_metadata_for_dataset(dataset_id):
+  def fetch_metadata_for_dataset(dataset_id, logger=False):
     """Download and return metadata.
 
     Parameters
@@ -20,14 +20,15 @@ def fetch_metadata_for_dataset(dataset_id):
       The metadata object.
     """
     url = Network.METADATA_URL_TEMPLATE % dataset_id
-    print(f'fetching metadata at {url}')
+    if logger:
+      logger.info('fetching metadata', url=url)
     df = Network.fetch_dataset(url)
     df["Update Date"] = pandas.to_datetime(df["Update Date"])
     df.sort_values("Update Date", inplace=True)
     df.set_index("Update Date", inplace=True)
     return df
 
-  def fetch_dataset(url, pandas_impl=pandas):
+  def fetch_dataset(url, pandas_impl=pandas, logger=False):
     """Download and return a dataset.
 
     Type inference is disabled in favor of explicit type casting at the
@@ -44,6 +45,6 @@ def fetch_dataset(url, pandas_impl=pandas):
     pandas.DataFrame
       The dataset.
     """
-
-    print(f'fetching dataset at {url}')
+    if logger:
+      logger.info('fetching dataset', url=url)
     return pandas_impl.read_csv(url, dtype=str)

src/acquisition/covid_hosp/common/utils.py

@@ -6,6 +6,8 @@
 
 import pandas as pd
 
+from delphi.epidata.acquisition.common.logger import get_structured_logger
+
 class CovidHospException(Exception):
   """Exception raised exclusively by `covid_hosp` utilities."""
 
@@ -69,7 +71,15 @@ def parse_bool(value):
       return False
     raise CovidHospException(f'cannot convert "{value}" to bool')
 
-  def issues_to_fetch(metadata, newer_than, older_than):
+  def limited_string_fn(length):
+    def limited_string(value):
+      value = str(value)
+      if len(value) > length:
+        raise CovidHospException(f"Value '{value}':{len(value)} longer than max {length}")
+      return value
+    return limited_string
+
+  def issues_to_fetch(metadata, newer_than, older_than, logger=False):
     """
     Construct all issue dates and URLs to be ingested based on metadata.
 
@@ -81,6 +91,8 @@ def issues_to_fetch(metadata, newer_than, older_than):
       Lower bound (exclusive) of days to get issues for.
     older_than Date
       Upper bound (exclusive) of days to get issues for
+    logger structlog.Logger [optional; default False]
+      Logger to receive messages
     Returns
     -------
       Dictionary of {issue day: list of (download urls, index)}
@@ -100,11 +112,12 @@ def issues_to_fetch(metadata, newer_than, older_than):
       elif day >= older_than:
         n_beyond += 1
     if n_beyond > 0:
-      print(f"{n_beyond} issues available on {older_than} or newer")
+      if logger:
+        logger.info("issues available", on_or_newer=older_than, count=n_beyond)
     return daily_issues
 
   @staticmethod
-  def merge_by_key_cols(dfs, key_cols):
+  def merge_by_key_cols(dfs, key_cols, logger=False):
     """Merge a list of data frames as a series of updates.
 
     Parameters:
@@ -113,13 +126,20 @@ def merge_by_key_cols(dfs, key_cols):
         Data frames to merge, ordered from earliest to latest.
       key_cols: list(str)
         Columns to use as the index.
+      logger structlog.Logger [optional; default False]
+        Logger to receive messages
 
     Returns a single data frame containing the most recent data for each state+date.
     """
 
     dfs = [df.set_index(key_cols) for df in dfs
            if not all(k in df.index.names for k in key_cols)]
     result = dfs[0]
+    if logger and len(dfs) > 7:
+      logger.warning(
+        "expensive operation",
+        msg="concatenating more than 7 files may result in long running times",
+        count=len(dfs))
     for df in dfs[1:]:
       # update values for existing keys
       result.update(df)
@@ -153,22 +173,25 @@ def update_dataset(database, network, newer_than=None, older_than=None):
     bool
       Whether a new dataset was acquired.
     """
-    metadata = network.fetch_metadata()
+    logger = get_structured_logger(f"{database.__class__.__module__}.{database.__class__.__name__}.update_dataset")
+    
+    metadata = network.fetch_metadata(logger=logger)
     datasets = []
     with database.connect() as db:
-      max_issue = db.get_max_issue()
+      max_issue = db.get_max_issue(logger=logger)
 
     older_than = datetime.datetime.today().date() if newer_than is None else older_than
     newer_than = max_issue if newer_than is None else newer_than
-    daily_issues = Utils.issues_to_fetch(metadata, newer_than, older_than)
+    daily_issues = Utils.issues_to_fetch(metadata, newer_than, older_than, logger=logger)
     if not daily_issues:
-      print("no new issues, nothing to do")
+      logger.info("no new issues; nothing to do")
       return False
     for issue, revisions in daily_issues.items():
       issue_int = int(issue.strftime("%Y%m%d"))
       # 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)
+                                        db.KEY_COLS,
+                                        logger=logger)
       # add metadata to the database
       all_metadata = []
       for url, index in revisions:
@@ -180,10 +203,10 @@ def update_dataset(database, network, newer_than=None, older_than=None):
       ))
     with database.connect() as db:
       for issue_int, dataset, all_metadata in datasets:
-        db.insert_dataset(issue_int, dataset)
+        db.insert_dataset(issue_int, dataset, logger=logger)
         for url, metadata_json in all_metadata:
-          db.insert_metadata(issue_int, url, metadata_json)
-        print(f'successfully acquired {len(dataset)} rows')
+          db.insert_metadata(issue_int, url, metadata_json, logger=logger)
+        logger.info("acquired rows", count=len(dataset))
 
       # note that the transaction is committed by exiting the `with` block
       return True

src/acquisition/covid_hosp/facility/database.py

@@ -40,7 +40,7 @@ class Database(BaseDatabase):
       Columndef('ccn', 'ccn', str),
       Columndef('city', 'city', str),
       Columndef('fips_code', 'fips_code', str),
-      Columndef('geocoded_hospital_address', 'geocoded_hospital_address', str),
+      Columndef('geocoded_hospital_address', 'geocoded_hospital_address', Utils.limited_string_fn(32)),
       Columndef('hhs_ids', 'hhs_ids', str),
       Columndef('hospital_name', 'hospital_name', str),
       Columndef('hospital_subtype', 'hospital_subtype', str),