[saanikaaa]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces an automated import pipeline for the EPA US Air Pollutant Emission Trends (County) dataset, including a download and preprocessing script, metadata configurations, mapping files, and test datasets. The review feedback focuses on enhancing the pipeline's robustness and correctness: preventing potential deadlocks in subprocess execution, ensuring both generated datasets are specified in the manifest's import inputs, downloading files atomically via temporary paths, and explicitly verifying that the output directory is not empty.

[gemini-code-assist]

Using subprocess.Popen with stderr=subprocess.PIPE without reading from proc.stderr can lead to a deadlock if the child process writes enough data to fill the OS pipe buffer. Since we do not need to capture the stderr output, redirecting it to subprocess.DEVNULL avoids this risk. Additionally, critical failure points in pipeline scripts should raise exceptions to ensure the process exits with a non-zero status code.

        proc = subprocess.Popen(
            ["mdb-export", db_path, table],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            encoding="utf-8",
            errors="replace",
        )

        with open(out_file, "w", newline="", encoding="utf-8") as f_out:
            reader = csv.reader(proc.stdout)
            writer = csv.writer(f_out)

            header = next(reader, None)
            if header is not None:
                writer.writerow(["year"] + header)
                stfips_idx = header.index("stfips") if "stfips" in header else None
                state_idx = header.index("state") if "state" in header else None
                for row in reader:
                    if row:
                        if (
                            year == 2014
                            and stfips_idx is not None
                            and state_idx is not None
                            and len(row) > max(stfips_idx, state_idx)
                            and row[state_idx].strip() == "Tribal"
                        ):
                            row[stfips_idx] = "88"
                        writer.writerow([year] + row)

        ret_code = proc.wait()
        if ret_code != 0:
            raise RuntimeError(f"mdb-export failed with exit code {ret_code} for table {table}")

References

1. 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.

[gemini-code-assist]

The current configuration only lists output_files/county_tier1.tmcf as the template_mcf, but the pipeline runs two separate imports producing two different TMCF files: county_tier1.tmcf and county_tier1_aggregate.tmcf. To ensure both datasets are correctly imported, both template/CSV pairs should be explicitly listed in the import_inputs array.

            "import_inputs": [
                {
                    "template_mcf": "output_files/county_tier1.tmcf",
                    "cleaned_csv": "output_files/county_tier1.csv"
                },
                {
                    "template_mcf": "output_files/county_tier1_aggregate.tmcf",
                    "cleaned_csv": "output_files/county_tier1_aggregate.csv"
                }
            ],

[gemini-code-assist]

To prevent leaving partially downloaded or corrupted files in the destination directory if the download is interrupted or fails, it is safer to stream the download to a temporary file first and then atomically rename it to the final destination path once the download is fully complete.

Suggested change

	def download_file(url: str, dest_path: str):
	"""Streams the download to destination path."""
	logging.info(f"Downloading: {url} -> {os.path.basename(dest_path)}")
	headers = {"User-Agent": "Mozilla/5.0"}
	with requests.get(url, headers=headers, stream=True, timeout=120) as resp:
	resp.raise_for_status()
	with open(dest_path, "wb") as f:
	for chunk in resp.iter_content(chunk_size=1024 * 1024):
	if chunk:
	f.write(chunk)
	def download_file(url: str, dest_path: str):
	"""Streams the download to destination path."""
	logging.info(f"Downloading: {url} -> {os.path.basename(dest_path)}")
	headers = {"User-Agent": "Mozilla/5.0"}
	temp_dest = dest_path + ".tmp"
	try:
	with requests.get(url, headers=headers, stream=True, timeout=120) as resp:
	resp.raise_for_status()
	with open(temp_dest, "wb") as f:
	for chunk in resp.iter_content(chunk_size=1024 * 1024):
	if chunk:
	f.write(chunk)
	os.replace(temp_dest, dest_path)
	finally:
	if os.path.exists(temp_dest):
	os.remove(temp_dest)

[gemini-code-assist]

Per the repository's general rules, when verifying the presence of input data in a directory, we should explicitly check if the directory is empty (using not os.listdir(...)) in addition to checking its existence. If empty, raise a FileNotFoundError and use a bare raise in the except block to ensure descriptive error messages are logged and the original exception is propagated.

Suggested change

	# 4. Move all county CSV files to county_input_file subfolder
	check_and_create_dir(COUNTY_OUTPUT_DIR)
	for fname in os.listdir(OUTPUT_DIR):
	src_path = os.path.join(OUTPUT_DIR, fname)
	if (
	os.path.isfile(src_path)
	and fname.lower().endswith(".csv")
	and "county" in fname.lower()
	):
	dest_path = os.path.join(COUNTY_OUTPUT_DIR, fname)
	logging.info(f"Moving county CSV: {fname} -> {dest_path}")
	shutil.move(src_path, dest_path)
	# 4. Move all county CSV files to county_input_file subfolder
	check_and_create_dir(COUNTY_OUTPUT_DIR)
	for fname in os.listdir(OUTPUT_DIR):
	src_path = os.path.join(OUTPUT_DIR, fname)
	if (
	os.path.isfile(src_path)
	and fname.lower().endswith(".csv")
	and "county" in fname.lower()
	):
	dest_path = os.path.join(COUNTY_OUTPUT_DIR, fname)
	logging.info(f"Moving county CSV: {fname} -> {dest_path}")
	shutil.move(src_path, dest_path)
	try:
	if not os.listdir(COUNTY_OUTPUT_DIR):
	raise FileNotFoundError(f"No county CSV files were found in {COUNTY_OUTPUT_DIR}")
	except FileNotFoundError:
	logging.error(f"Verification failed: {COUNTY_OUTPUT_DIR} is empty.")
	raise

References

1. When verifying the presence of input data in a directory, explicitly check if the directory is empty (e.g., using not os.listdir(...)) in addition to checking its existence. If empty, raise an appropriate exception (like FileNotFoundError) and use a bare raise in the except block to ensure descriptive error messages are logged and the original exception is propagated.