popoto.transfer.import_

popoto.transfer.import_

Import records from a JSON Lines export produced by :mod:popoto.transfer.

Keys are always preserved. Records are saved one at a time -- deliberately not through bulk_create -- because an external pipeline makes save() return the pipeline for every record and destroys per-record observability, which the reconciliation ledger depends on.

BATCH_SIZE = 500 module-attribute

Records per conflict-check / reconciliation batch.

import_records(model_class, stream, on_conflict='error', on_write_gate='reject', on_embedding_mismatch='error')

Import records from a JSON Lines export into model_class.

Keys are always preserved, so re-running an import converges rather than duplicating, and every Relationship value and application-level key pointer keeps pointing at the same record.

Parameters:

Name	Type	Description	Default
model_class		The destination Model class.	required
stream	TextIO	A text file-like object positioned at the manifest line.	required
on_conflict	str	What to do when the destination already holds a key. "error" (default) refuses -- the only mode that cannot clobber; "skip" leaves the existing record untouched; "overwrite" replaces it, which makes a re-run idempotent.	'error'
on_write_gate	str	"reject" (default) honors the destination model's WriteFilterMixin gate and reports every refusal; "bypass" writes around it. Bypass only disables Popoto's own gate -- an application's save() override returning falsy still surfaces as a rejection.	'reject'
on_embedding_mismatch	str	"error" (default) refuses when the export's provider fingerprint differs from the destination's; "carry" imports the vectors anyway; "regenerate" drops them so on_save re-embeds.	'error'

Returns:

Name	Type	Description
An	ImportReport	class:ImportReport accounting for every record line as landed,
	ImportReport	skipped, rejected, errored, or partial, with a reason per non-landed
	ImportReport	record.

Raises:

Type	Description
ValueError	If a policy argument is not one of its allowed values.
ModelException	If the manifest is missing, the format version is unsupported, the model name does not match, an embedding provenance mismatch is refused, or on_conflict="error" hits a collision. The first three raise before anything is written.

Note

Import is not atomic across records and assumes the destination is not under concurrent write for this model. The recovery path for an interrupted run is to re-run with on_conflict="overwrite".

Example

with open("memories.jsonl") as fh: report = Memory.import_records(fh, on_conflict="overwrite") print(report.summary())

Source code in src/popoto/transfer/import_.py

def import_records(
    model_class,
    stream: TextIO,
    on_conflict: str = "error",
    on_write_gate: str = "reject",
    on_embedding_mismatch: str = "error",
) -> ImportReport:
    """Import records from a JSON Lines export into ``model_class``.

    Keys are always preserved, so re-running an import converges rather than
    duplicating, and every ``Relationship`` value and application-level key
    pointer keeps pointing at the same record.

    Args:
        model_class: The destination Model class.
        stream: A text file-like object positioned at the manifest line.
        on_conflict: What to do when the destination already holds a key.
            ``"error"`` (default) refuses -- the only mode that cannot
            clobber; ``"skip"`` leaves the existing record untouched;
            ``"overwrite"`` replaces it, which makes a re-run idempotent.
        on_write_gate: ``"reject"`` (default) honors the destination model's
            ``WriteFilterMixin`` gate and reports every refusal;
            ``"bypass"`` writes around it. Bypass only disables Popoto's own
            gate -- an application's ``save()`` override returning falsy still
            surfaces as a rejection.
        on_embedding_mismatch: ``"error"`` (default) refuses when the export's
            provider fingerprint differs from the destination's; ``"carry"``
            imports the vectors anyway; ``"regenerate"`` drops them so
            ``on_save`` re-embeds.

    Returns:
        An :class:`ImportReport` accounting for every record line as landed,
        skipped, rejected, errored, or partial, with a reason per non-landed
        record.

    Raises:
        ValueError: If a policy argument is not one of its allowed values.
        ModelException: If the manifest is missing, the format version is
            unsupported, the model name does not match, an embedding
            provenance mismatch is refused, or ``on_conflict="error"`` hits a
            collision. The first three raise before anything is written.

    Note:
        Import is not atomic across records and assumes the destination is not
        under concurrent write for this model. The recovery path for an
        interrupted run is to re-run with ``on_conflict="overwrite"``.

    Example:
        with open("memories.jsonl") as fh:
            report = Memory.import_records(fh, on_conflict="overwrite")
        print(report.summary())
    """
    _check_choice("on_conflict", on_conflict, _ON_CONFLICT)
    _check_choice("on_write_gate", on_write_gate, _ON_WRITE_GATE)
    _check_choice(
        "on_embedding_mismatch", on_embedding_mismatch, _ON_EMBEDDING_MISMATCH
    )

    report = ImportReport(model=model_class.__name__)

    lines = iter_lines(stream)
    manifest = None
    for _line_number, raw in lines:
        try:
            manifest = parse_line(raw)
        except ValueError as exc:
            raise ModelException(
                f"first line of the export is not valid JSON: {exc}"
            ) from exc
        break

    manifest = _validate_manifest(model_class, manifest)
    report.source_matched_count = manifest.get("matched_count")
    report.fidelity = {
        **(manifest.get("fields") or {}),
        **(manifest.get("mixins") or {}),
    }
    drop_state = _resolve_embedding_provenance(
        model_class, manifest, on_embedding_mismatch, report
    )

    batch: "list[dict]" = []
    for line_number, raw in lines:
        try:
            record = parse_line(raw)
        except ValueError as exc:
            report.add(f"line {line_number}", ERRORED, f"malformed JSON line: {exc}")
            continue
        key = record.get("key")
        if not isinstance(key, str) or not key:
            report.add(
                f"line {line_number}",
                ERRORED,
                "record has no usable 'key'; keys are always preserved, so a "
                "record without one cannot be placed",
            )
            continue
        batch.append(record)
        if len(batch) >= BATCH_SIZE:
            _process_batch(
                model_class, batch, report, on_conflict, on_write_gate, drop_state
            )
            batch = []

    if batch:
        _process_batch(
            model_class, batch, report, on_conflict, on_write_gate, drop_state
        )

    return report