ditto.recorders¶

def recorder_from_files(
    *,
    save: Callable[[T, Path], None],
    load: Callable[[Path], T],
    suffix: str,
) -> Recorder[T]:
    """
    Adapt a library that reads and writes files to the byte-based `Recorder`.

    Each `dumps` and `loads` call uses a fresh temporary directory holding one
    file, `snapshot<suffix>`, which is removed afterwards whether the call
    succeeds or fails. Libraries that can serialise to bytes directly should
    build a `Recorder` from their in-memory functions instead, avoiding the
    file I/O.

    Only single-file formats are supported. A value returned by `load` must not
    depend on the temporary file after `load` returns, so lazy or memory-mapped
    readers are unsuitable.

    Parameters
    ----------
    save : Callable[[T, Path], None]
        Function that writes a value to the given path.
    load : Callable[[Path], T]
        Function that reads a value from the given path.
    suffix : str
        Filename suffix the library expects, e.g. `".parquet"`. It names only
        the temporary file; the snapshot's filename comes from the name the
        recorder is registered under.

    Returns
    -------
    Recorder[T]
        A recorder whose `dumps` and `loads` go through a temporary file.

    Raises
    ------
    ValueError
        If `suffix` does not start with `.`, is only `.`, or contains a path
        separator.
    """
    if not suffix.startswith(".") or suffix == "." or "/" in suffix or "\\" in suffix:
        raise ValueError(
            f"suffix must be a filename suffix such as '.parquet', got {suffix!r}"
        )

    def dumps(value: T) -> bytes:
        with TemporaryDirectory() as directory:
            path = Path(directory) / f"snapshot{suffix}"
            save(value, path)
            return path.read_bytes()

    def loads(raw: bytes) -> T:
        with TemporaryDirectory() as directory:
            path = Path(directory) / f"snapshot{suffix}"
            path.write_bytes(raw)
            return load(path)

    return Recorder(dumps=dumps, loads=loads)

def register(
    name: str,
    recorder: Recorder,
    registry: RecorderRegistry = RECORDER_REGISTRY,
) -> None:
    """
    Add a recorder to the given registry under a new name.

    The name is also the recorder's persisted identifier: it ends the
    recorder's snapshot filenames and is recorded in `ditto.lock`.

    Parameters
    ----------
    name : str
        Name to register the recorder under, following the recorder name
        grammar (`<format>` or `<namespace>.<format>`).
    recorder : Recorder
        The recorder instance to register.
    registry : RecorderRegistry, optional
        Registry to add to. Defaults to the shared `RECORDER_REGISTRY`.
        Pass an isolated `RecorderRegistry([], [])` in tests to avoid mutating
        shared state.

    Raises
    ------
    DittoRecorderConflictError
        If `name` is already registered, or breaks another naming rule together
        with the names already registered. Registered recorders cannot be
        replaced.
    """
    registry.register(name, recorder)

def get(
    name: str,
    registry: Mapping[str, Recorder] = RECORDER_REGISTRY,
    fallback: Recorder | object = _MISSING,
) -> Recorder:
    """
    Look up a recorder by name.

    Parameters
    ----------
    name : str
        Key to look up in the registry.
    registry : Mapping[str, Recorder], optional
        Registry to query. Defaults to the shared `RECORDER_REGISTRY`.
        Pass an isolated dict in tests to avoid depending on shared state.
    fallback : Recorder, optional
        Recorder to return when `name` is not found. If omitted, an unknown
        recorder raises `DittoUnknownRecorderError`. A registered recorder that
        fails to load raises `DittoRecorderLoadError`, even if a fallback is given.

    Returns
    -------
    Recorder
        The registered recorder, or an explicitly supplied `fallback`.
    """
    if name in registry:
        return registry[name]
    if fallback is not _MISSING:
        return cast(Recorder, fallback)
    raise DittoUnknownRecorderError(name, list(registry))

def default() -> Recorder:
    """
    Return the default recorder.

    Returns
    -------
    Recorder
        The strict JSON recorder.
    """
    return _default