Skip to content

cdm_data_loaders.pipelines.jsonlines.extract_jsonschema_validate_pipeline

Load JSONL records validated against an external JSON Schema registry.

The module selected by entity_schemas must define ENTITY_SCHEMAS: a mapping of table names to JSON Schema dictionaries, each declaring $schema. Each validated JSON Schema is dereferenced and converted to a pyarrow.Schema via JSONSchemaToPyArrow. Files are read from one input subdirectory per table. Valid records are coerced to the target pyarrow.Schema and buffered into pyarrow.Table pages via DictBuffer; parsing and validation failures go to _rejected with table_name, source_file, line_no, raw_record, and error_detail. Format checks are enabled.

build_entity_resource 

build_entity_resource(
    table_name: str,
    schema: ValidatedSchema,
    settings: JsonlJsonschemaIngestSettings,
) -> DltResource

Read plain or gzip JSONL files for one entity and route validated pages.

A missing entity directory produces no rows. Validated records are coerced to the target PyArrow schema(s) and buffered into pyarrow.Table pages via DictBuffer. Nested values are preserved rather than normalized into child tables.

Parameters:

Name Type Description Default
table_name str

table name

 required
schema ValidatedSchema

validated JSON Schema for the table

 required
settings JsonlJsonschemaIngestSettings

pipeline configuration

 required

Returns:

Type Description
DltResource

resource yielding validated and rejected records for this entity
Source code in src/cdm_data_loaders/pipelines/jsonlines/extract_jsonschema_validate_pipeline.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def build_entity_resource(
    table_name: str,
    schema: ValidatedSchema,
    settings: JsonlJsonschemaIngestSettings,
) -> DltResource:
    """Read plain or gzip JSONL files for one entity and route validated pages.

    A missing entity directory produces no rows. Validated records are coerced
    to the target PyArrow schema(s) and buffered into ``pyarrow.Table`` pages via
    ``DictBuffer``. Nested values are preserved rather than normalized into
    child tables.

    :param table_name: table name
    :type  table_name: str
    :param schema: validated JSON Schema for the table
    :type  schema: ValidatedSchema
    :param settings: pipeline configuration
    :type  settings: JsonlJsonschemaIngestSettings
    :return: resource yielding validated and rejected records for this entity
    :rtype: DltResource
    """
    return build_shared_entity_resource(
        table_name,
        settings,
        resolve_error=_make_resolve_error(schema),
    )

run_jsonlines_pipeline 

run_jsonlines_pipeline(
    settings: JsonlJsonschemaIngestSettings,
) -> LoadInfo | None

Load requested tables using the validated registry and precomputed PyArrow schemas.

Parameters:

Name Type Description Default
settings JsonlJsonschemaIngestSettings

pipeline configuration

 required

Returns:

Type Description
LoadInfo | None

load information for the pipeline
Source code in src/cdm_data_loaders/pipelines/jsonlines/extract_jsonschema_validate_pipeline.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def run_jsonlines_pipeline(settings: JsonlJsonschemaIngestSettings) -> LoadInfo | None:
    """Load requested tables using the validated registry and precomputed PyArrow schemas.

    :param settings: pipeline configuration
    :type settings:  JsonlJsonschemaIngestSettings
    :return: load information for the pipeline
    :rtype:  LoadInfo | None
    """
    pipeline_name = str(settings.model_config.get("cli_prog_name") or JSONSCHEMA_PIPELINE_NAME)
    return run_registry_ingest(
        settings,
        registry=settings.schema_registry.registry,
        build_resource=build_entity_resource,
        pipeline_name=pipeline_name,
    )