cdm_data_loaders.readers.jsonlines ¶

def iter_file_jsonl_records(file_item: FileItemDict) -> Generator[LineRecord]:
    """Open ``file_item`` and yield parsed ``LineRecord`` dicts for each non-blank line.

    dlt's default ``compression="auto"`` decompresses files that the listing marked as gzip, which it
    does from the file name (``*.gz``) for every filesystem.
    """
    with file_item.open(mode="rt", encoding="utf-8") as fh:
        for line_no, line in enumerate(fh, start=1):
            stripped = line.strip()
            if not stripped:
                continue
            parse_error: str | None = None
            record: dict[str, Any] | None = None
            try:
                record = json.loads(stripped)
            except json.JSONDecodeError as e:
                parse_error = str(e)
            yield {
                "record": record,
                "raw_record": stripped,
                "parse_error": parse_error,
                "source_file": file_item["relative_path"],
                "line_no": line_no,
            }

def stream_jsonl_lines(items: Iterator[FileItemDict], buffer_size: int) -> Generator[TDataItems, Any, Any]:
    """Read each file in items. Yield pages of line records.

    Each record has: record (parsed JSON, or None on parse failure),
    raw_record, parse_error (None on success), source_file, line_no.
    Records are accumulated into pages of ``buffer_size`` before being
    yielded, so downstream transformers receive pages rather than
    individual rows.

    :param items: file items to read
    :type items: Iterator[FileItemDict]
    :param buffer_size: number of line records per yielded page
    :type buffer_size: int
    :yield: pages of line records
    :rtype: Generator[TDataItems, Any, Any]
    """
    page: list[LineRecord] = []
    for file_item in items:
        for line_record in iter_file_jsonl_records(file_item):
            page.append(line_record)
            if len(page) >= buffer_size:
                yield page
                page = []
    if page:
        yield page

def make_page_router(  # noqa: PLR0913
    table_name: str,
    buffer_size: int,
    resolve_error: ParseErrorResolver | None = None,
    *,
    transform_valid: Callable[[Any], Any] | None = None,
    table_transform: RecordTransformFn | None = None,
    schema_map: ArrowSchemaRegistry | None = None,
    use_pyarrow: bool = True,
) -> Callable[[Iterable[dict[str, Any]]], Generator[Any, Any, Any]]:
    """Build a function that routes pages of line records to valid and rejected tables.

    For each item, ``resolve_error`` (when provided) inspects the record and
    returns None for a valid record, or the error detail string otherwise.
    Parse errors bypass ``resolve_error`` entirely. Valid records yield
    ``item["record"]`` passed through ``transform_valid`` (identity when None) and
    ``table_transform`` into target table(s); failed records yield a provenance
    row (``table_name``, ``source_file``, ``line_no``, ``raw_record``,
    ``error_detail``) into ``_rejected``. Records are accumulated in a
    ``DictBuffer``, which emits ``_rejected`` and schematized target tables as
    ``pyarrow.Table`` pages when ``use_pyarrow`` is ``True``, or as ``list[dict]``
    pages otherwise.

    :param table_name: default table name for valid records
    :type  table_name: str
    :param buffer_size: number of rows to buffer per table before yielding a page
    :type  buffer_size: int
    :param resolve_error: optional function returning None for valid records or error detail
    :type  resolve_error: ParseErrorResolver | None
    :param transform_valid: function applied to each valid record before routing
    :type  transform_valid: Callable[[Any], Any] | None
    :param table_transform: optional transform function for filtering, transforming, or splitting records
    :type  table_transform: RecordTransformFn | None
    :param schema_map: optional per-table PyArrow schema registry
    :type  schema_map: ArrowSchemaRegistry | None
    :param use_pyarrow: whether buffered pages are emitted as ``pyarrow.Table`` instances
    :type  use_pyarrow: bool
    :return: function taking one page (or iterable) of line records and yielding routed pages
    :rtype : Callable[[Iterable[dict[str, Any]]], Generator[Any, Any, Any]]
    """

    def _route(page: Iterable[dict[str, Any]] | dict[str, Any] | pa.Table) -> Generator[Any, Any, Any]:
        if isinstance(page, pa.Table):
            yield dlt.mark.with_table_name(page, REJECTED_TABLE_NAME)
            return
        if isinstance(page, dict):
            page = [page]
        buffer = DictBuffer(max_items=buffer_size, schema_map=schema_map, use_pyarrow=use_pyarrow)
        for item in page:
            if "parse_error" not in item and "error_detail" in item:
                yield from buffer.add_item(REJECTED_TABLE_NAME, item)
                continue
            yield from _route_line_record(
                item,
                table_name,
                buffer,
                resolve_error=resolve_error,
                transform_valid=transform_valid,
                table_transform=table_transform,
                schema_map=schema_map,
            )
        yield from buffer.flush()

    return _route

def route_jsonl_lines(
    items: Iterator[FileItemDict],
    settings: JsonlExtractSettings,
) -> Generator[TDataItems, Any, Any]:
    """Read each file in items and route each line by its file's table name.

    Valid records yield into the file's table with the given name (or transformed
    tables when ``table_transform`` is supplied).
    Records that fail JSON parsing yield into ``_rejected`` with
    ``table_name``, ``source_file``, ``line_no``, ``raw_record``, and ``error_detail``.

    :param items: file items to read
    :type items: Iterator[FileItemDict]
    :param settings: pipeline configuration
    :type  settings: JsonlExtractSettings
    :yield: table-tagged rows
    :rtype: Generator[TDataItems, Any, Any]
    """
    buffer = DictBuffer.from_settings(settings)
    for file_item in items:
        for item in iter_file_jsonl_records(file_item):
            yield from _route_line_record(
                item,
                settings.table_name,
                buffer,
                table_transform=settings.get_table_transform(settings.table_name),
                schema_map=settings.schema_map,
            )
    yield from buffer.flush()