Software Engineer

6 minute read

10 min read 2178 words

In the healthcare data world, the National Provider Identifier (NPI) dataset from CMS is essential. However, the files are massive, and the manual process of checking for updates, downloading, and unzipping them is a chore.

In this post, we’ll walk through a professional-grade Python script designed to automate the entire pipeline: scraping the CMS NPPES page, downloading the latest ZIP files with a retry strategy, and extracting them for immediate use.

1. Key Features of the Script

This isn’t just a simple “fetch and save” script. It incorporates several production-ready patterns:

  • Resilient Networking: Uses a custom retry strategy to handle transient network errors (500, 502, 503, 504).
  • Memory Efficient: Streams large files to disk in chunks instead of loading them into RAM.
  • Automated Log Rotation: Keeps logs tidy by rotating them daily and retaining only the last 30 days.
  • Clean Environment: Automatically clears the target directory before starting to ensure a fresh state (configurable).
  • Post-Download Processing: Unzips every downloaded file into its own dedicated subdirectory.

2. Setting Up Your Environment

Before running the script, ensure you have the necessary libraries installed. We recommend using a virtual environment.

# Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install requests beautifulsoup4 python-dotenv

Configuration via .env

The script supports environment variables for easy configuration without touching the code. Create a .env file in your project root:

CMS_URL=https://download.cms.gov/nppes/NPI_Files.html
TARGET_DIR=./npi_data
LOG_FILE=npi_pipeline.log

3. Deep Dive into the Implementation

3.1 Resilient Sessions

We use requests.Session coupled with HTTPAdapter and Retry. This ensures that if CMS’s servers are momentarily busy (returning a 429 or 503), our script doesn’t just crash; it waits and retries with an exponential backoff.

3.2 Streaming for Large Data

NPI files can be several gigabytes. By setting stream=True in our get request and iterating over chunk_size=8192, we keep our memory footprint constant regardless of the file size.

3.3 The NPIDownloader Orchestrator

The script uses an Object-Oriented approach. The NPIDownloader class handles the state (base URL, session, directory) and provides clean methods for each step of the process (_get_zip_links, _download_file, _unzip_file).

4. The Full Script

Save the following code as download_npi.py:

"""A script to scrape and download NPI files from CMS.

This script identifies ZIP files on the CMS NPPES page, downloads them to a
target directory, and extracts their contents.

Author: Thomas Ngo
Date: 2026-07-17
"""

import argparse
import logging
from logging.handlers import TimedRotatingFileHandler
import os
import shutil
import time
import zipfile
from pathlib import Path
from typing import List
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Default Configuration
DEFAULT_CMS_URL = "https://download.cms.gov/nppes/NPI_Files.html"
DEFAULT_TARGET_DIR = "./npi_downloads"
DEFAULT_LOG_FILE = "download_npi.log"

logger = logging.getLogger(__name__)


class NPIDownloader:
    """Handles scraping, downloading, and extracting NPI files from CMS."""

    def __init__(self, base_url: str, download_dir: Path):
        """Initializes the downloader with a base URL and target directory."""
        self.base_url = base_url
        self.download_dir = download_dir
        self.session = self._create_session()

    def _create_session(self) -> requests.Session:
        """Creates a requests Session with a retry strategy."""
        session = requests.Session()
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        session.mount("http://", adapter)
        return session

    def clear_target_directory(self):
        """Deletes all files and subdirectories in the target directory."""
        logger.info("Clearing directory: %s", self.download_dir)

        if not self.download_dir.exists():
            return

        for item in self.download_dir.iterdir():
            try:
                if item.is_file() or item.is_symlink():
                    item.unlink()
                elif item.is_dir():
                    shutil.rmtree(item)
            except Exception as e:
                logger.error("Failed to delete %s: %s", item, e)

    def _get_zip_links(self) -> List[str]:
        """Scrapes the base URL for links ending in .zip."""
        logger.info("Fetching page: %s", self.base_url)
        try:
            response = self.session.get(self.base_url, timeout=30)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            logger.error("Failed to connect to CMS: %s", e)
            return []

        soup = BeautifulSoup(response.text, 'html.parser')
        links = soup.find_all('a', href=True)
        return [l['href'] for l in links if l['href'].endswith('.zip')]

    def _download_file(self, url: str, save_path: Path, retries: int = 3) -> bool:
        """Downloads a file using streaming."""
        for attempt in range(retries):
            try:
                logger.info("Downloading: %s (Attempt %d/%d)", save_path.name, attempt + 1, retries)
                with self.session.get(url, stream=True, timeout=60) as r:
                    r.raise_for_status()
                    with open(save_path, 'wb') as f:
                        for chunk in r.iter_content(chunk_size=8192):
                            if chunk:
                                f.write(chunk)
                return True
            except Exception as e:
                logger.error("Error downloading %s: %s", url, e)
                if attempt < retries - 1:
                    time.sleep(2 ** (attempt + 1))
        return False

    def _unzip_file(self, zip_path: Path, extract_to: Path) -> bool:
        """Unzips a ZIP file to the specified directory."""
        try:
            logger.info("Unzipping: %s", zip_path.name)
            extract_to.mkdir(parents=True, exist_ok=True)
            with zipfile.ZipFile(zip_path, 'r') as zip_ref:
                zip_ref.extractall(extract_to)
            return True
        except Exception as e:
            logger.error("Error unzipping %s: %s", zip_path, e)
            return False

    def run(self):
        """Orchestrates the download and extraction process."""
        self.download_dir.mkdir(parents=True, exist_ok=True)
        zip_links = self._get_zip_links()
        
        if not zip_links:
            logger.warning("No ZIP files found.")
            return

        for link in zip_links:
            file_url = urljoin(self.base_url, link)
            file_name = Path(link).name
            file_path = self.download_dir / file_name

            if self._download_file(file_url, file_path):
                extract_path = self.download_dir / file_path.stem
                self._unzip_file(file_path, extract_path)


def setup_logging(log_file: str):
    """Configures rotating file logging."""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            TimedRotatingFileHandler(log_file, when="D", interval=1, backupCount=30),
            logging.StreamHandler()
        ],
        force=True
    )


def main():
    load_dotenv()
    parser = argparse.ArgumentParser(description="Scrape and download NPI files from CMS NPPES.")
    parser.add_argument("--url", default=os.getenv("CMS_URL", DEFAULT_CMS_URL))
    parser.add_argument("--dir", default=os.getenv("TARGET_DIR", DEFAULT_TARGET_DIR))
    parser.add_argument("--log", default=os.getenv("LOG_FILE", DEFAULT_LOG_FILE))
    parser.add_argument("--no-clear", action="store_true")

    args = parser.parse_args()
    setup_logging(args.log)

    target_path = Path(args.dir)
    downloader = NPIDownloader(args.url, target_path)

    if not args.no_clear:
        downloader.clear_target_directory()

    downloader.run()


if __name__ == "__main__":
    main()

5. Running the Script

You can run the script using the default settings:

python download_npi.py

Or customize the behavior using command-line arguments:

# Change the target directory and skip clearing existing files
python download_npi.py --dir ./my_data --no-clear

6. Conclusion

Automating data ingestion from public sources like CMS requires more than just a GET request. By implementing retries, streaming, and proper logging, you create a pipeline that is reliable and easy to maintain. This script provides a solid foundation for any data engineer looking to synchronize NPI data for their applications.

Additional notes

Scheduling on Windows

You can use the Windows Task Scheduler to run the script automatically on weekends or every week:

  1. Open Task Scheduler: Press Win + S, search for “Task Scheduler”, and open it.
  2. Create Basic Task: Click Create Basic Task… in the “Actions” pane on the right.
  3. Name the Task: Provide a name (e.g., NPI Downloader) and click Next.
  4. Task Trigger:
    • Select Weekly and click Next.
    • Set the Start time and select the days (e.g., Saturday and Sunday for weekends). Click Next.
  5. Action: Select Start a program and click Next.
  6. Settings:
    • Program/script: Enter the full path to your virtual environment’s Python executable. Example: C:\Users\YourName\download-dds-file\env\Scripts\python.exe
    • Add arguments: Enter download_npi.py.
    • Start in: Enter the full path to your project directory. Example: C:\Users\YourName\download-dds-file
  7. Finish: Review the settings and click Finish.

Tip: To avoid issues with relative paths, always use absolute paths for “Program/script” and “Start in”.

Integration with SQL Server Agent

You can integrate this script into SQL Server Agent to automate it within your database environment:

  1. Open SSMS: Open SQL Server Management Studio and connect to your instance.
  2. Create New Job: Expand SQL Server Agent, right-click Jobs, and select New Job….
  3. General: Name the job (e.g., NPI_Downloader_Job).
  4. Steps:
    • Click New… to add a job step.
    • Step name: Run Python Script.
    • Type: Select Operating system (CmdExec).
    • Command: Enter the full path to the Python executable and the script, enclosed in quotes if they contain spaces. Example: "C:\Users\YourName\download-dds-file\env\Scripts\python.exe" "C:\Users\YourName\download-dds-file\download_npi.py"
    • Click OK.
  5. Schedules: Click New… to define when the script should run (e.g., Weekly, Daily).
  6. Finish: Click OK to create the job.

Permissions Note: The account running the SQL Server Agent service must have read/write permissions to the project directory and the target download directory.

Leave a comment

You may also enjoy