Skip to content

io

io #

Analysis-side disk verbs: load a captured value, commit one, copy file paths.

commit(path, value, schema=None) #

Write value rooted at path; unchanged leaves are left alone, none are removed.

Parameters:

Name Type Description Default
path Path | str

Destination root.

required
value object

The value to write.

required
schema type[CGType] | dict[str, JSONValue] | None

The type to store value as; defaults to its own type, required for a bare Path or to port onto a different type.

None

Returns:

Type Description
Path

The destination Path.

Source code in capturegraph-lib/capturegraph/types/io.py
def commit(
    path: Path | str,
    value: object,
    schema: type[CGType] | dict[str, JSONValue] | None = None,
) -> Path:
    """Write ``value`` rooted at ``path``; unchanged leaves are left alone, none are removed.

    Args:
        path: Destination root.
        value: The value to write.
        schema: The type to store ``value`` as; defaults to its own type, required
            for a bare ``Path`` or to port onto a different type.

    Returns:
        The destination ``Path``.
    """
    cgtype = _commit_type(value, schema)
    destination = Path(path)
    cgtype.store(destination, value)
    return destination

copy(*, src, dst) #

Copy files from source paths to destination paths.

Destination parent directories are created as needed. A missing or None source produces a Missing placeholder in the result instead of raising.

Parameters:

Name Type Description Default
src Array

Array of source file paths.

required
dst Array

Array of destination paths (same length as src).

required

Returns:

Type Description
Array

Array of destination paths that were created.

Source code in capturegraph-lib/capturegraph/types/io.py
def copy(*, src: Array, dst: Array) -> Array:
    """Copy files from source paths to destination paths.

    Destination parent directories are created as needed. A missing or None
    source produces a Missing placeholder in the result instead of raising.

    Args:
        src: Array of source file paths.
        dst: Array of destination paths (same length as src).

    Returns:
        Array of destination paths that were created.
    """

    def _copy_one(src_path: object, dst_path: Path) -> Path | _Missing:
        if is_missing(src_path) or src_path is None:
            return _Missing(ValueError(f"Source path is missing: {src_path}"))
        if not isinstance(src_path, (str, Path)):
            return _Missing(TypeError(f"Source path is not a path: {src_path!r}"))
        dst_path = Path(dst_path)
        dst_path.parent.mkdir(parents=True, exist_ok=True)
        copy2(src_path, dst_path)
        return dst_path

    return Array(_copy_one(s, d) for s, d in zip(src, dst, strict=True))

load(path, schema) #

load(path: Path | str, schema: type[T]) -> T
load(
    path: Path | str, schema: dict[str, JSONValue]
) -> object

Load a captured value from path according to schema.

File leaves resolve to their path (the media is opened only on demand), so a load is metadata-weight; to read one slot of a large target without the rest, point path and schema at the sub-tree.

Parameters:

Name Type Description Default
path Path | str

The capture root on disk.

required
schema type[CGType] | dict[str, JSONValue]

The type describing the data — a cg.Struct subclass (or any CGType), or a schema dict (rebuilt with cg.from_schema).

required

Returns:

Type Description
object

The loaded value — a Struct/Array/Map that broadcasts

object

attribute access for vectorized analysis.

Source code in capturegraph-lib/capturegraph/types/io.py
def load(path: Path | str, schema: type[CGType] | dict[str, JSONValue]) -> object:
    """Load a captured value from ``path`` according to ``schema``.

    File leaves resolve to their path (the media is opened only on demand), so a
    load is metadata-weight; to read one slot of a large target without the rest,
    point ``path`` and ``schema`` at the sub-tree.

    Args:
        path: The capture root on disk.
        schema: The type describing the data — a ``cg.Struct`` subclass (or any
            ``CGType``), or a schema ``dict`` (rebuilt with ``cg.from_schema``).

    Returns:
        The loaded value — a ``Struct``/``Array``/``Map`` that broadcasts
        attribute access for vectorized analysis.
    """
    cgtype = from_schema(schema) if isinstance(schema, dict) else schema
    return cgtype.load(Path(path))