cdm_data_loaders.pipelines.jsonlines.extract_pydantic_validate_pipeline ¶

JSONL validate-and-load pipeline for the KBase CTS.

Reads JSONL files. Validates each record against a Pydantic model. Writes valid and invalid records to separate tables.

The entity models come from an external module set by entity_models. That module must define:

ENTITY_MODELS: dict[str, type[BaseModel]]

Each key is a table name. Each value is the Pydantic model for that table. Example:

from datetime import date
from pydantic import BaseModel

class Sample(BaseModel):
    id: str
    collection_date: date

ENTITY_MODELS: dict[str, type[BaseModel]] = {"sample": Sample}

input_dir must contain one subdirectory per table name in ENTITY_MODELS. Each subdirectory holds that table's JSONL files. Files may be plain text or gzip-compressed (.jsonl or .jsonl.gz).

Valid records go to a table named after the entity. Records that fail JSON parsing or model validation go to the _rejected table. Each rejected record includes the target table name, source file, line number, raw text, and error detail. Write disposition is append or replace only. No merge is performed.

build_entity_resource ¶

build_entity_resource(
    table_name: str,
    model: type[BaseModel],
    settings: JsonlPydanticIngestSettings,
) -> DltResource

Build the resource that reads, validates, and routes one entity's records.

Reads files matching settings.file_glob from settings.input_dir/table_name. If that directory does not exist, uses an empty file list instead. A missing entity then produces zero rows, not an error.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	entity name; also the input_dir subdirectory name	required
`model`	`type[BaseModel]`	Pydantic model used to validate this entity's records	required
`settings`	`JsonlPydanticIngestSettings`	pipeline settings	required

Returns:

Type	Description
`DltResource`	resource yielding validated and rejected records for this entity

Source code in src/cdm_data_loaders/pipelines/jsonlines/extract_pydantic_validate_pipeline.py

def build_entity_resource(
    table_name: str, model: type[BaseModel], settings: JsonlPydanticIngestSettings
) -> DltResource:
    """Build the resource that reads, validates, and routes one entity's records.

    Reads files matching `settings.file_glob` from
    `settings.input_dir/table_name`. If that directory does not exist,
    uses an empty file list instead. A missing entity then produces zero
    rows, not an error.

    :param table_name: entity name; also the input_dir subdirectory name
    :type  table_name: str
    :param model: Pydantic model used to validate this entity's records
    :type  model: type[BaseModel]
    :param settings: pipeline settings
    :type  settings: JsonlPydanticIngestSettings
    :return: resource yielding validated and rejected records for this entity
    :rtype: DltResource
    """
    resolve_error, dump_validated = _make_validator_callbacks(model)

    return build_shared_entity_resource(
        table_name,
        settings,
        resolve_error=resolve_error,
        transform_valid=dump_validated,
    )

run_jsonlines_with_validation_pipeline ¶

run_jsonlines_with_validation_pipeline(
    settings: JsonlPydanticIngestSettings,
) -> LoadInfo | None

Run the JSONL validate-and-load pipeline for every requested table.

Parameters:

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

Returns:

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

Source code in src/cdm_data_loaders/pipelines/jsonlines/extract_pydantic_validate_pipeline.py

def run_jsonlines_with_validation_pipeline(settings: JsonlPydanticIngestSettings) -> LoadInfo | None:
    """Run the JSONL validate-and-load pipeline for every requested table.

    :param settings: pipeline configuration
    :type  settings: JsonlPydanticIngestSettings
    :return: load information for the pipeline
    :rtype: LoadInfo | None
    """
    return run_registry_ingest(
        settings,
        registry=settings.model_registry.registry,
        build_resource=build_entity_resource,
        pipeline_name=PYDANTIC_PIPELINE_NAME,
    )