cdm_data_loaders.pipelines.xsv.pipeline ¶

XSV (CSV/TSV) ingestion pipeline for the KBase CTS.

Cleans and validates each XSV file against settings.schema_file using qsv (see readers.jsonschema_xsv.xsv_validator), then loads each valid row, cast to the types declared in the schema, into table_name. Files that fail cleaning/validation are routed to _rejected, one row per recorded error; qsv's own per-row validation diagnostics are left in its scratch output directory and are not currently surfaced as rows.

XsvWorkPaths `dataclass` ¶

Scratch resources shared across all files processed in one pipeline run.

Parameters:

Name	Type	Description	Default
`qsv_cmd`	`str`	path to the qsv binary	required
`header_file_path`	`Path`	header file used when an input file is missing a header row	required
`tmp_dir`	`Path`	scratch directory for intermediate cleaning/validation files	required
`qsv_output_dir`	`Path`	directory qsv writes failure diagnostics to	required
`validated_dir`	`Path`	directory the final cleaned, validated file is written to	required
`download_dir`	`Path`	directory non-local input files are downloaded to before validation (qsv needs local files)	required

Source code in src/cdm_data_loaders/pipelines/xsv/pipeline.py

@dataclass
class XsvWorkPaths:
    """Scratch resources shared across all files processed in one pipeline run.

    :param qsv_cmd: path to the qsv binary
    :type qsv_cmd: str
    :param header_file_path: header file used when an input file is missing a header row
    :type header_file_path: Path
    :param tmp_dir: scratch directory for intermediate cleaning/validation files
    :type tmp_dir: Path
    :param qsv_output_dir: directory qsv writes failure diagnostics to
    :type qsv_output_dir: Path
    :param validated_dir: directory the final cleaned, validated file is written to
    :type validated_dir: Path
    :param download_dir: directory non-local input files are downloaded to before validation (qsv needs local files)
    :type download_dir: Path
    """

    qsv_cmd: str
    header_file_path: Path
    tmp_dir: Path
    qsv_output_dir: Path
    validated_dir: Path
    download_dir: Path

run_xsv_pipeline ¶

run_xsv_pipeline(
    settings: XsvIngestSettings,
) -> LoadInfo | None

Run the XSV ingestion pipeline on matching files in settings.input_dir.

Parameters:

Name	Type	Description	Default
`settings`	`XsvIngestSettings`	pipeline configuration	required

Returns:

Type	Description
`LoadInfo \| None`	load information for the pipeline

Source code in src/cdm_data_loaders/pipelines/xsv/pipeline.py

def run_xsv_pipeline(settings: XsvIngestSettings) -> LoadInfo | None:
    """Run the XSV ingestion pipeline on matching files in settings.input_dir.

    :param settings: pipeline configuration
    :type  settings: XsvIngestSettings
    :return: load information for the pipeline
    :rtype: LoadInfo | None
    """
    qsv_cmd = qsv_check()

    with (
        TemporaryDirectory(prefix="xsv_tmp_") as tmp_dir,
        TemporaryDirectory(prefix="xsv_qsv_output_") as qsv_output_dir,
        TemporaryDirectory(prefix="xsv_validated_") as validated_dir,
        TemporaryDirectory(prefix="xsv_download_") as download_dir,
    ):
        work_paths = XsvWorkPaths(
            qsv_cmd=qsv_cmd,
            header_file_path=generate_header(
                settings.validated_schema,
                Path(tmp_dir),
                delimiter=settings.parsing_config.get("delimiter", "\t"),
            ),
            tmp_dir=Path(tmp_dir),
            qsv_output_dir=Path(qsv_output_dir),
            validated_dir=Path(validated_dir),
            download_dir=Path(download_dir),
        )

        xsv_resource = build_per_file_filesystem_resource(
            settings,
            _process_xsv_file,
            work_paths,
            name="xsv_reader",
            schema_contract=DEFAULT_SCHEMA_CONTRACT,
        )

        return run_pipeline(
            settings=settings,
            resource=xsv_resource,
            **build_file_ingest_kwargs(settings, PIPELINE_NAME),
        )

cdm_data_loaders.pipelines.xsv.pipeline ¶

XsvWorkPaths dataclass ¶

run_xsv_pipeline ¶

XsvWorkPaths `dataclass` ¶