typedecide.data.loaders

Reading decisions from JSONL, JSON, CSV/TSV, Parquet and Hugging Face datasets. Task-oriented walkthrough: Bring your own data.

typedecide.data.loaders

Reading decisions out of whatever the customer already has.

This is the first surface a new user touches, and the data they bring is a CSV a ticketing system exported at four in the afternoon. So every failure here names the row, names the column, quotes the offending value and says what would fix it. A loader that says "invalid input" has made the library someone else's problem.

SUFFIX_FORMATS module-attribute

SUFFIX_FORMATS: dict[str, str] = {
    ".jsonl": "jsonl",
    ".ndjson": "jsonl",
    ".json": "json",
    ".csv": "csv",
    ".tsv": "csv",
    ".parquet": "parquet",
    ".pq": "parquet",
}

FORMATS module-attribute

FORMATS: tuple[str, ...] = (
    "jsonl",
    "json",
    "csv",
    "parquet",
    "hf",
)

OPTION_DELIMITERS module-attribute

OPTION_DELIMITERS: tuple[str, ...] = (
    "||",
    "|",
    ";",
    "\t",
    ",",
)

Resolver module-attribute

Resolver: TypeAlias = Callable[[str | None], str | None]

FieldMapping dataclass

FieldMapping(
    state: str = "state",
    question: str = "question",
    options: str = "options",
    answer: str = "answer_id",
    group: str | None = None,
    id: str | None = None,
    option_ids: str = "option_ids",
    key: str = "key",
    option_delimiter: str | None = None,
)

Maps a customer's column names onto our schema, so bring-your-own-data does not mean rename-your-columns-first.

The first six fields are the contract. The rest exist because CSV has to express a set of options in a flat table, and there is no one spelling of that. group and id default to None, which means "use the conventional column if the file happens to have one, otherwise leave it unset".

state class-attribute instance-attribute

state: str = 'state'

question class-attribute instance-attribute

question: str = 'question'

options class-attribute instance-attribute

options: str = 'options'

answer class-attribute instance-attribute

answer: str = 'answer_id'

group class-attribute instance-attribute

group: str | None = None

id class-attribute instance-attribute

id: str | None = None

option_ids class-attribute instance-attribute

option_ids: str = 'option_ids'

key class-attribute instance-attribute

key: str = 'key'

option_delimiter class-attribute instance-attribute

option_delimiter: str | None = None

load_decisions

load_decisions(
    source: str | Path,
    *,
    format: str | None = None,
    mapping: FieldMapping | None = None
) -> list[Decision]

Load decisions from a file or a Hugging Face dataset id.

format is inferred from the suffix when not given: .jsonl/.ndjson, .json, .csv/.tsv, .parquet/.pq. A string that is not an existing path and looks like owner/name is read as a HF dataset id.

Source code in src/typedecide/data/loaders.py

def load_decisions(
    source: str | Path,
    *,
    format: str | None = None,  # noqa: A002 - the contract names this argument
    mapping: FieldMapping | None = None,
) -> list[Decision]:
    """Load decisions from a file or a Hugging Face dataset id.

    ``format`` is inferred from the suffix when not given: ``.jsonl``/``.ndjson``,
    ``.json``, ``.csv``/``.tsv``, ``.parquet``/``.pq``. A string that is not an
    existing path and looks like ``owner/name`` is read as a HF dataset id.
    """
    mapping = mapping or FieldMapping()
    resolved = _resolve_format(source, format)

    if resolved == "hf":
        return _load_hf(str(source), mapping)

    path = Path(source)
    if not path.exists():
        raise DataError(
            f"No such dataset file: {path.resolve()}. "
            "Fix: check the path, or pass a Hugging Face dataset id such as 'owner/name'."
        )
    if path.is_dir():
        raise DataError(
            f"{path.resolve()} is a directory, not a dataset file. "
            "Fix: name the .jsonl, .json, .csv or .parquet file inside it."
        )

    loaders: dict[str, Callable[[Path, FieldMapping], list[Decision]]] = {
        "jsonl": _load_jsonl,
        "json": _load_json,
        "csv": _load_csv,
        "parquet": _load_parquet,
    }
    loader = loaders.get(resolved)
    if loader is None:  # pragma: no cover - _resolve_format only returns names in FORMATS
        raise DataError(f"Unknown format {resolved!r}. Fix: pass one of {', '.join(FORMATS)}.")
    try:
        return loader(path, mapping)
    except UnicodeDecodeError as error:
        raise DataError(
            f"{path.name} is not UTF-8: byte 0x{error.object[error.start]:02x} at offset "
            f"{error.start} does not decode. Fix: re-export the file as UTF-8 (in Excel, "
            "'CSV UTF-8 (Comma delimited)'), or convert it with "
            "`iconv -f WINDOWS-1252 -t UTF-8`."
        ) from error
    except OSError as error:
        raise DataError(
            f"Could not read {path}: {error}. Fix: check the file's permissions."
        ) from error

write_decisions

write_decisions(
    decisions: Sequence[Decision], path: Path
) -> None

Write decisions back out. .json gets an array, anything else gets JSONL.

Source code in src/typedecide/data/loaders.py

def write_decisions(decisions: Sequence[Decision], path: Path) -> None:
    """Write decisions back out. ``.json`` gets an array, anything else gets JSONL."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)

    # Written beside the target and renamed over it, so a crash or a full disk leaves
    # the previous file intact rather than a truncated dataset that still parses.
    # newline="\n" keeps the bytes identical on Windows, where text mode would
    # otherwise write "\r\n" and change the file's hash from one machine to the next.
    scratch = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        with scratch.open("w", encoding="utf-8", newline="\n") as handle:
            if path.suffix.lower() == ".json":
                payloads = [decision_to_dict(d) for d in decisions]
                handle.write(json.dumps(payloads, ensure_ascii=False, indent=2, default=str))
                handle.write("\n")
            else:
                # One row at a time: JSONL never needs the whole dataset serialised at once.
                for decision in decisions:
                    handle.write(
                        json.dumps(decision_to_dict(decision), ensure_ascii=False, default=str)
                    )
                    handle.write("\n")
        scratch.replace(path)
    except OSError as error:
        raise DataError(f"Could not write {path}: {error}.") from error
    finally:
        with contextlib.suppress(OSError):
            scratch.unlink()
    LOGGER.info("wrote %d decisions to %s", len(decisions), path)