INNER CODE UNIT · Python

chunk

i3T4AN/KADATH · kadath/engine.py:47

            chunk = stream.read(1024 * 1024)
            if not chunk: break
            size += len(chunk); hasher.update(chunk)
    return size, hasher.hexdigest()


def atomic_write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + ".tmp-" + uuid.uuid4().hex)
    encoded = json.dumps(payload, indent=2, sort_keys=True, default=str)
    with temporary.open("w") as stream:
        stream.write(encoded); stream.flush(); os.fsync(stream.fileno())
    temporary.replace(path)


def scoped_regular_bytes(root: Path, item: Path) -> bytes:
    """Read a regular file without following organism-created symlinks."""
    root = root.absolute(); item = item.absolute(); relative = item.relative_to(root)

View source record →