cdm_data_loaders.pipelines.jsonlines.base ¶

Shared machinery for the JSONL registry-ingest pipelines.

Each pipeline reads one JSONL input subdirectory per table, routes valid records to a table named after the entity, and routes failures to _rejected with provenance columns (table_name, source_file, line_no, raw_record, error_detail). This module holds the pieces all registry pipelines share; each pipeline module supplies only its validator and registry loader.

build_entity_resource ¶

build_entity_resource(
    table_name: str,
    settings: Any,
    resolve_error: ParseErrorResolver,
    *,
    transform_valid: Callable[[Any], Any] | None = None
) -> DltResource

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

A missing entity directory produces no rows. Nested values are preserved rather than normalized into child tables.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	table name; also the input_dir subdirectory name	required
`settings`	`Any`	pipeline configuration with input_dir, file_glob, buffer_size	required
`resolve_error`	`ParseErrorResolver`	function returning None for valid records or error detail	required
`transform_valid`	`Callable[[Any], Any] \| None`	validation function applied to each record before yielding	`None`

Returns:

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

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

def build_entity_resource(
    table_name: str,
    settings: Any,  # noqa: ANN401
    resolve_error: ParseErrorResolver,
    *,
    transform_valid: Callable[[Any], Any] | None = None,
) -> DltResource:
    """Read plain or gzip JSONL files for one entity and route validated pages.

    A missing entity directory produces no rows. Nested values are preserved
    rather than normalized into child tables.

    :param table_name: table name; also the input_dir subdirectory name
    :type  table_name: str
    :param settings: pipeline configuration with input_dir, file_glob, buffer_size
    :type  settings: Any
    :param resolve_error: function returning None for valid records or error detail
    :type  resolve_error: ParseErrorResolver
    :param transform_valid: validation function applied to each record before yielding
    :type  transform_valid: Callable[[Any], Any] | None
    :return: resource yielding validated and rejected records for this entity
    :rtype: DltResource
    """
    raw = build_filesystem_resource(
        settings,
        _stream_entity_lines,
        name=f"{table_name}_raw",
        file_glob=f"{table_name}/{settings.file_glob}",
        table_name=table_name,
        max_table_nesting=0,
    )
    validated = dlt.transformer(
        make_page_router(
            table_name,
            settings.buffer_size,
            resolve_error,
            transform_valid=transform_valid,
            table_transform=settings.get_table_transform(table_name),
            schema_map=settings.schema_map,
            use_pyarrow=settings.use_pyarrow,
        ),
        data_from=raw,
        name=f"{table_name}_validated",
        max_table_nesting=0,
        parallelized=True,
    )
    validated.apply_hints(schema_contract=TSchemaContractDict(**DEFAULT_SCHEMA_CONTRACT))  # pyright: ignore[reportCallIssue]
    return validated

run_registry_ingest ¶

run_registry_ingest(
    settings: Any,
    registry: dict[
        RegistryKey, type[BaseModel] | ValidatedSchema
    ],
    build_resource: ResourceFactory,
    pipeline_name: str,
    pipeline_run_kwargs: dict[str, Any] | None = None,
) -> LoadInfo | None

Load tables using the shared pipeline runner.

Parameters:

Name	Type	Description	Default
`settings`	`Any`	pipeline configuration	required
`registry`	`dict[RegistryKey, type[BaseModel] \| ValidatedSchema]`	registry of table name to its schema/model entry	required
`build_resource`	`ResourceFactory`	function building one entity's resource	required
`pipeline_name`	`str`	registered pipeline name	required
`pipeline_run_kwargs`	`dict[str, Any] \| None`	extra kwargs for pipeline.run	`None`

Returns:

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

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

def run_registry_ingest(
    settings: Any,  # noqa: ANN401
    registry: dict[RegistryKey, type[BaseModel] | ValidatedSchema],
    build_resource: ResourceFactory,
    pipeline_name: str,
    pipeline_run_kwargs: dict[str, Any] | None = None,
) -> LoadInfo | None:
    """Load tables using the shared pipeline runner.

    :param settings: pipeline configuration
    :type  settings: Any
    :param registry: registry of table name to its schema/model entry
    :type  registry: dict[RegistryKey, type[BaseModel] | ValidatedSchema]
    :param build_resource: function building one entity's resource
    :type  build_resource: ResourceFactory
    :param pipeline_name: registered pipeline name
    :type  pipeline_name: str
    :param pipeline_run_kwargs: extra kwargs for pipeline.run
    :type  pipeline_run_kwargs: dict[str, Any] | None
    :return: load information for the pipeline
    :rtype:  LoadInfo | None
    """
    resources = [build_resource(table_name, registry[table_name], settings) for table_name in sorted(registry)]
    ingest_kwargs = build_file_ingest_kwargs(settings, pipeline_name, fallback_dataset_name=True)
    if pipeline_run_kwargs:
        ingest_kwargs["pipeline_run_kwargs"].update(pipeline_run_kwargs)
    return run_pipeline(
        settings=settings,
        resource=resources,
        **ingest_kwargs,
    )