Automate CDC_Mortality_UnderlyingCause_SingleRace import - #2240
shvngisingh wants to merge 1 commit into
Conversation
- Add download.py to automate live data downloading from CDC WONDER (D158) with session agreement handling, state-level querying, and dynamic 2-year chunking to respect the 75k row export limit. - Add download_test.py with unit tests covering query construction, session initialization, 75k row cap partitioning, large state chunking, and 429 rate-limiting backoff. - Update download.sh to execute download.py. - Update README.md to reflect automated import pipeline and document download parameters and execution commands.
shvngisingh
changed the title
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request automates the CDC WONDER Single Race Mortality Data download process by introducing a Python-based live downloader (download.py), updating the shell wrapper, and adding unit tests. Key feedback includes making the is_state_downloaded check more robust to verify coverage of all requested years (and updating the tests accordingly), avoiding the suppression of exceptions when re-initializing sessions during retries to prevent confusing downstream errors, and correcting the cron expression in the README since standard cron evaluates day-of-month and day-of-week as an OR condition.
| if years: | ||
| latest_year = years[-1] | ||
| has_chunk = any( | ||
| f.name.endswith(f"_{latest_year}.csv") | ||
| or f"_{latest_year}_" in f.name for f in matches) | ||
| if has_chunk: | ||
| return True | ||
| single_file = Path( | ||
| output_dir) / f"UnderlyingCauseofDeath_SingleRace_{state_fips}.csv" | ||
| if single_file.exists(): | ||
| content = single_file.read_text(encoding="utf-8", errors="replace") | ||
| return f",{latest_year}," in content | ||
| return False |
There was a problem hiding this comment.
The current implementation of is_state_downloaded only checks if the latest year in the requested list is present in the output directory. If a file for the latest year exists (or a chunk containing it), the function returns True and skips downloading the state entirely, even if other requested years are missing from the output directory.
A more robust approach is to verify that every requested year is covered by either the "all" file or the chunk files.
if years:
covered_years = set()
for f in matches:
name = f.stem
parts = name.split("_")
if len(parts) == 3:
try:
content = f.read_text(encoding="utf-8", errors="replace")
for y in years:
if f",{y}," in content:
covered_years.add(y)
except Exception:
pass
elif len(parts) == 4:
covered_years.add(parts[3])
elif len(parts) == 5:
try:
start, end = int(parts[3]), int(parts[4])
for y in range(start, end + 1):
covered_years.add(str(y))
except ValueError:
pass
return all(y in covered_years for y in years)References
- It is acceptable to silently skip invalid data entries if they are not considered errors and logging them would create unnecessary noise.
| This import pipeline is configured to run Semi-automatic on the second Saturday of every month schedule. | ||
| This import pipeline is configured to run automatically on the second Saturday of every month schedule. | ||
|
|
||
| - Cron Expression: 30 08 8-14 * 6 |
There was a problem hiding this comment.
In standard cron, if both the day-of-month (8-14) and day-of-week (6) are specified, they are evaluated as an OR condition rather than an AND condition. This means the job will run on every day from the 8th to the 14th of the month, as well as on every Saturday of the month.
To restrict execution to only the second Saturday of the month, you should schedule the cron job to run every Saturday (e.g., 30 08 * * 6) and add a shell check in the command to verify if the day of the month is between 8 and 14:
30 08 * * 6 [ $(date +\%d) -ge 8 ] && [ $(date +\%d) -le 14 ] && sh download.sh| def test_is_state_downloaded(self): | ||
| with tempfile.TemporaryDirectory() as temp_dir: | ||
| self.assertFalse(download.is_state_downloaded(temp_dir, "02")) | ||
|
|
||
| # Create empty file | ||
| f = Path(temp_dir) / "UnderlyingCauseofDeath_SingleRace_02.csv" | ||
| f.write_text("") | ||
| self.assertFalse(download.is_state_downloaded(temp_dir, "02")) | ||
|
|
||
| # Create valid file > 100 bytes | ||
| f.write_text("Header,col1,col2,col3\n" + | ||
| "val1,val2,val3,val4\n" * 10) | ||
| self.assertTrue(download.is_state_downloaded(temp_dir, "02")) |
There was a problem hiding this comment.
Update the unit test for is_state_downloaded to cover the new robust logic checking that all requested years are covered by either the "all" file or the chunk files.
def test_is_state_downloaded(self):
with tempfile.TemporaryDirectory() as temp_dir:
self.assertFalse(download.is_state_downloaded(temp_dir, "02"))
# Create empty file
f = Path(temp_dir) / "UnderlyingCauseofDeath_SingleRace_02.csv"
f.write_text("")
self.assertFalse(download.is_state_downloaded(temp_dir, "02"))
# Create valid file > 100 bytes
f.write_text("Header,col1,col2,col3\n" +
",2018,val\n,2019,val\n" * 10)
self.assertTrue(download.is_state_downloaded(temp_dir, "02"))
self.assertTrue(download.is_state_downloaded(temp_dir, "02", ["2018", "2019"]))
self.assertFalse(download.is_state_downloaded(temp_dir, "02", ["2018", "2020"]))
# Test chunk files
f.unlink()
f_chunk = Path(temp_dir) / "UnderlyingCauseofDeath_SingleRace_02_2018_2019.csv"
f_chunk.write_text("Header,col1,col2,col3\n" + "val1,val2,val3,val4\n" * 10)
self.assertTrue(download.is_state_downloaded(temp_dir, "02", ["2018", "2019"]))
self.assertFalse(download.is_state_downloaded(temp_dir, "02", ["2018", "2020"]))| try: | ||
| self.init_session() | ||
| payload = self._build_post_data(state_fips, years) | ||
| except Exception as e: | ||
| logging.warning("Session re-initialization error: %s", | ||
| e) |
There was a problem hiding this comment.
Suppressing the exception when self.init_session() fails during a 429 retry will cause the loop to continue to the next iteration with self.action_url set to None. This leads to a confusing ValueError or MissingSchema exception on the subsequent self.session.post call. Letting the exception propagate naturally allows the outer retry mechanism to handle it correctly.
self.init_session()
payload = self._build_post_data(state_fips, years)References
- In pipeline scripts, critical failure points should raise exceptions to ensure the process exits with a non-zero status code, rather than returning False and allowing the script to exit with success.
| try: | ||
| self.init_session() | ||
| payload = self._build_post_data(state_fips, years) | ||
| except Exception as session_err: | ||
| logging.warning("Session re-initialization error: %s", | ||
| session_err) |
There was a problem hiding this comment.
Suppressing the exception when self.init_session() fails during standard retry recovery will cause the loop to continue to the next iteration with self.action_url set to None. This leads to a confusing ValueError or MissingSchema exception on the subsequent self.session.post call. Letting the exception propagate naturally allows the outer retry mechanism to handle it correctly.
self.init_session()
payload = self._build_post_data(state_fips, years)References
- In pipeline scripts, critical failure points should raise exceptions to ensure the process exits with a non-zero status code, rather than returning False and allowing the script to exit with success.
1 participant
Automate CDC_Mortality_UnderlyingCause_SingleRace Data Acquisition
Overview
This PR transitions the CDC Single Race Underlying Cause of Death import pipeline (
statvar_imports/us_cdc/single_race/) from a Semi-Automated workflow (requiring manual web downloads via CDC WONDER form dropdowns and GCS staging copies) to a Fully Automated live data acquisition pipeline.download.pyto directly negotiate CDC WONDER sessions, agree to data use terms, and extract county-level mortality statistics programmatically.gs://datcom-prod-imports/...) down to the exact byte (78,468,106 bytes) and exact row count (813,880 observations across all 805 statistical variables with 0 differences).download_test.pywith 12 unit tests verifying session initialization, query parameter building, 75k-row limit partitioning, large state chunking, and HTTP 429 retry backoff.1. Key Modernization Changes
A. Live CDC WONDER Downloader (
download.py)https://wonder.cdc.gov/ucd-icd10-expanded.html(Database D158: Underlying Cause of Death, Single Race).action-I Agree) and extracts active session tokens and hidden form fields.LARGE_STATESset: CA, TX, NY, FL, etc.) are preemptively partitioned into 2-year chunks (e.g.,2018_2019,2020_2021,2022_2023,2024).Too Many Requests) with exponential backoff and Retry-After header compliance..tmpfiles, and renames atomically to prevent partial writes.B. Updated Shell Execution Wrapper (
download.sh)gsutil cp/gcloud storage cpcommands fromgs://unresolved_mcf/...with direct execution ofpython3 "${SCRIPT_DIR}/download.py" "$@".set -e -o pipefailand ensuresinput_files/directory creation.C. Unit Test Suite (
download_test.py)D. Documentation (
README.md)Semi-AutomatedtoAutomated.--states,--years,--delay,--batch_size), and execution workflows.2. Production Parity & Validation Audit
The output generated from the automated
download.pypipeline was cross-validated against the active production dataset in GCS:gs://datcom-prod-imports/statvar_imports/us_cdc/single_race/CDC_Mortality_UnderlyingCause_SingleRace/2026_08_15T01_32_55_219057_07_00/underlyingcauseofdeath2018_2023singlerace.csvgs://pulkeet-dc-bucket/goutam/single_race/output/underlyingcauseofdeath2018_2024singlerace.csvdownload.py)78,468,106 bytes78,468,106 bytes813,880 rows813,880 rowsAnnual Observation Breakdown (2018–2024):
3. Scope of Changes
statvar_imports/us_cdc/single_race/download.pystatvar_imports/us_cdc/single_race/download_test.pystatvar_imports/us_cdc/single_race/download.shdownload.py.statvar_imports/us_cdc/single_race/README.mdExclusions: No large raw data files (
input_files/), output artifacts (output/), cache databases (http_cache.sqlite), or temporary files are included in this PR.4. Verification & Testing
python3 -m unittest discover -s statvar_imports/us_cdc/single_race -p "*_test.py"