Validate CSV Files with Shell and Python Scripts in a Working Directory

id: csv-validation-shell-python
namespace: company.team
description: |
  Validate a CSV file in two passes inside one working directory: a bash
  script sanity-checks the file and counts rows, then a Python script loads
  it with pandas and verifies that required columns exist. Both scripts ship
  inline via inputFiles so the blueprint deploys standalone; in production,
  keep them as namespace files synced from Git and swap inputFiles for
  namespaceFiles.

tasks:
  - id: my_working_dir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared filesystem for the sample data and both validation scripts.
    tasks:
      - id: create_csv
        type: io.kestra.plugin.core.storage.Write
        description: Sample data file for the validation tasks below.
        content: |
          first_name,last_name,email,signup_date
          Ada,Lovelace,ada@example.com,1843-12-10
          Grace,Hopper,grace@example.com,1906-12-09
          Katherine,Johnson,kj@example.com,1918-08-26
        extension: ".csv"

      - id: shell_check
        type: io.kestra.plugin.scripts.shell.Commands
        description: Fast structural check, confirms the file exists and counts data rows.
        inputFiles:
          check.sh: |
            #!/bin/bash

            if [[ $# -ne 1 ]]; then
              echo "Usage: $0 /path/to/users.csv" >&2
              exit 1
            fi

            CSV_FILE="$1"
            if [[ ! -f "$CSV_FILE" ]]; then
              echo "Error: file not found: $CSV_FILE" >&2
              exit 1
            fi

            total_lines=$(wc -l < "$CSV_FILE" | tr -d ' ')
            if [[ "$total_lines" -le 1 ]]; then
              echo "No user rows found in $CSV_FILE"
              exit 0
            fi

            data_rows=$(( total_lines - 1 ))
            echo "File: $CSV_FILE"
            echo "Total lines: $total_lines (including header)"
            echo "User rows: $data_rows"

            echo
            echo "Header row:"
            head -n 1 "$CSV_FILE"
        commands:
          - bash check.sh "{{ outputs.create_csv.uri }}"

      - id: python_validate
        type: io.kestra.plugin.scripts.python.Commands
        description: Deeper validation, loads the CSV with pandas and fails if required
          columns are missing.
        containerImage: python:3.13-slim
        dependencies:
          - pandas
          - click
        inputFiles:
          pre_process.py: |
            """Utility for validating CSV inputs prior to Kestra processing."""

            from __future__ import annotations

            import logging
            import sys
            from pathlib import Path
            from typing import Sequence

            import click
            import pandas as pd


            logging.basicConfig(
                level=logging.INFO,
                format="%(levelname)s: %(message)s",
                stream=sys.stdout,
            )


            def pre_process(input_file: Path, column_names: Sequence[str]) -> None:
                """Load a CSV file, print quick stats, and ensure required columns exist."""

                logging.info(f"Reading CSV from {input_file}")
                dataframe = pd.read_csv(input_file)

                logging.info(f"CSV details - rows: {len(dataframe)}, columns: {len(dataframe.columns)}")
                logging.info(f"Available columns: {', '.join(dataframe.columns)}")

                missing_columns = [col for col in column_names if col not in dataframe.columns]
                if missing_columns:
                    logging.error(f"Missing required columns: {', '.join(missing_columns)}")
                    raise ValueError(f"CSV validation failed; missing columns: {', '.join(missing_columns)}")

                logging.info(f"Column validation successful for: {', '.join(column_names)}")


            @click.command()
            @click.option(
                "--input-file",
                "input_file",
                type=click.Path(exists=True, dir_okay=False, readable=True, path_type=Path),
                required=True,
                help="Path to the CSV file to validate.",
            )
            @click.option(
                "--column-names",
                "column_names",
                type=str,
                required=True,
                help="Comma-separated list of column names that must exist in the CSV.",
            )
            def main(input_file: Path, column_names: str) -> None:
                """CLI entry point for the pre-process validation script."""

                requested_columns = [col.strip() for col in column_names.split(",") if col.strip()]
                if not requested_columns:
                    raise click.BadParameter(
                        "At least one column name must be provided.",
                        param_hint="--column-names",
                    )

                try:
                    pre_process(input_file=input_file, column_names=requested_columns)
                except ValueError as exc:
                    raise click.ClickException(str(exc)) from exc


            if __name__ == "__main__":
                main()
        commands:
          - python3 pre_process.py --input-file="{{ outputs.create_csv.uri }}"
            --column-names="first_name,last_name,email"