added conditional for merging

Author: aysim319

claims_hosp/delphi_claims_hosp/backfill.py

@@ -12,6 +12,7 @@
 import shutil
 from datetime import datetime, timedelta
 from typing import Union
+import calendar
 
 # third party
 import pandas as pd
@@ -57,7 +58,7 @@ def store_backfill_file(claims_filepath, _end_date, backfill_dir, logger):
     backfilldata = backfilldata.loc[(backfilldata["time_value"] >= _start_date)
                                     & (~backfilldata["fips"].isnull()),
                                     selected_columns]
-    logger.info("Filtering backfill data", startdate=_start_date, enddate=_end_date)
+    logger.info("Filtering source data", startdate=_start_date, enddate=_end_date)
 
     backfilldata["lag"] = [(_end_date - x).days for x in backfilldata["time_value"]]
     backfilldata["time_value"] = backfilldata.time_value.dt.strftime("%Y-%m-%d")
@@ -76,9 +77,9 @@ def store_backfill_file(claims_filepath, _end_date, backfill_dir, logger):
     # Store intermediate file into the backfill folder
     try:
         backfilldata.to_parquet(path, index=False)
-        logger.info("Stored backfill data in parquet", filename=filename)
+        logger.info("Stored source data in parquet", filename=filename)
     except:
-        logger.info("Failed to store backfill data in parquet", )
+        logger.info("Failed to store source data in parquet")
     return path
 
 
@@ -162,8 +163,9 @@ def get_date(file_link):
 
     date_list = list(map(get_date, new_files))
     latest_date = max(date_list)
-    if latest_date.month == most_recent.month:
-        logger.info("Not a new month; skipping merging")
+    num_of_days_in_month = calendar.monthrange(latest_date.year, latest_date.month)[1]
+    if len(date_list) < num_of_days_in_month:
+        logger.info("Not enough days, skipping merging", n_file_days=len(date_list))
         return
 
     logger.info(f"Merging files", start_date=date_list[0], end_date=date_list[-1])

claims_hosp/tests/test_backfill.py

@@ -1,3 +1,4 @@
+import calendar
 import logging
 import os
 import glob
@@ -57,7 +58,7 @@ def test_store_backfill_file(self, caplog):
 
         self.cleanup()
 
-    def test_merge_backfill_file(self, caplog):
+    def test_merge_backfill_file(self, caplog, monkeypatch):
         fn = "claims_hosp_202006.parquet"
         caplog.set_level(logging.INFO)
         logger = get_structured_logger()
@@ -76,6 +77,7 @@ def test_merge_backfill_file(self, caplog):
             store_backfill_file(DATA_FILEPATH, dropdate, backfill_dir, logger)
 
         today = datetime(2020, 7, 1)
+        monkeypatch.setattr(calendar, 'monthrange', lambda x, y: (1, 4))
         merge_backfill_file(backfill_dir, today, logger,
                             test_mode=True)
         assert "Merging files" in caplog.text
@@ -102,6 +104,29 @@ def test_merge_backfill_file(self, caplog):
 
         self.cleanup()
 
+    def test_merge_backfill_file_no_call(self, caplog):
+        fn = "claims_hosp_202006.parquet"
+        caplog.set_level(logging.INFO)
+        logger = get_structured_logger()
+
+        # Check when there is no daily file to merge.
+        today = datetime(2020, 6, 14)
+        merge_backfill_file(backfill_dir, today, logger,
+                            test_mode=True)
+        assert fn not in os.listdir(backfill_dir)
+        assert "No new files to merge; skipping merging" in caplog.text
+
+        # Generate backfill daily files
+        for d in range(11, 15):
+            dropdate = datetime(2020, 6, d)
+            store_backfill_file(DATA_FILEPATH, dropdate, backfill_dir, logger)
+
+        today = datetime(2020, 7, 1)
+        merge_backfill_file(backfill_dir, today, logger,
+                            test_mode=True)
+        assert "Not enough days, skipping merging" in caplog.text
+        self.cleanup()
+
     def test_merge_existing_backfill_files(self, caplog):
         issue_date = datetime(year=2020, month=6, day=13)
         issue_date_str = issue_date.strftime("%Y%m%d")
@@ -164,8 +189,8 @@ def prep_backfill_data():
         file_to_add = store_backfill_file(DATA_FILEPATH, issue_date, backfill_dir, logger)
         merge_existing_backfill_files(backfill_dir, file_to_add, issue_date, logger)
         assert "Issue date has no matching merged files" in caplog.text
-
         self.cleanup()
 
 
 
+