hubvault.storage.pack

Append-only pack-file storage helpers for hubvault.storage.

This module provides a small pack-file abstraction used by the repository backend to persist chunk payload bytes outside the immutable JSON object store.

The module contains:

PackChunkLocation - Physical location of one chunk inside a pack
PackWriteResult - Result metadata for a completed pack write
PackStore - Reader and writer for append-only pack files

Example:

>>> from pathlib import Path
>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     store = PackStore(Path(tmpdir))
...     result = store.write_pack("demo", [b"abc", b"def"])
...     store.read_chunk(result.chunks[0])
b'abc'

PACK_MAGIC

hubvault.storage.pack.PACK_MAGIC = b'hubvault-pack/v1\n'

bytes(iterable_of_ints) -> bytes bytes(string, encoding[, errors]) -> bytes bytes(bytes_or_buffer) -> immutable copy of bytes_or_buffer bytes(int) -> bytes object of size given by the parameter initialized with null bytes bytes() -> empty bytes object

Construct an immutable array of bytes from:

an iterable yielding integers in range(256)
a text string encoded using the specified encoding
any object implementing the buffer API.
an integer

PackChunkLocation

class hubvault.storage.pack.PackChunkLocation(pack_id: str, offset: int, stored_size: int, logical_size: int)[source]

Describe the stored location of one chunk inside a pack file.

Parameters:

pack_id (str) – Pack identifier without the file suffix
offset (int) – Absolute byte offset inside the pack file
stored_size (int) – Stored chunk size in bytes
logical_size (int) – Logical chunk size in bytes

Example:

>>> location = PackChunkLocation("demo", 16, 4, 4)
>>> location.pack_id
'demo'

PackWriteResult

class hubvault.storage.pack.PackWriteResult(pack_id: str, pack_path: str, total_size: int, chunks: Tuple[PackChunkLocation, ...])[source]

Summarize one completed pack write.

Parameters:

pack_id (str) – Pack identifier without the file suffix
pack_path (str) – Absolute pack path on disk
total_size (int) – Total pack size in bytes
chunks (Tuple[PackChunkLocation, ...]) – Ordered chunk locations written into the pack

Example:

>>> result = PackWriteResult("demo", "/tmp/demo.pack", 32, tuple())
>>> result.pack_id
'demo'

PackStore

class hubvault.storage.pack.PackStore(pack_dir: str | Path)[source]

Read and write append-only chunk pack files.

Parameters:: pack_dir (Union[str, pathlib.Path]) – Directory containing .pack files

Example:

>>> store = PackStore("/tmp/packs")
>>> store.pack_dir.name
'packs'

__init__(pack_dir: str | Path) → None[source]

Initialize the pack store.

Parameters:: pack_dir (Union[str, pathlib.Path]) – Directory containing .pack files
Returns:: None.
Return type:: None

Example:

>>> PackStore("/tmp/packs").pack_dir.name
'packs'

pack_path(pack_id: str) → Path[source]

Build the absolute path for a pack identifier.

Parameters:: pack_id (str) – Pack identifier without the file suffix
Returns:: Absolute pack path
Return type:: pathlib.Path

Example:

>>> PackStore("/tmp/packs").pack_path("demo").name
'demo.pack'

read_chunk(location: PackChunkLocation) → bytes[source]

Read one full chunk from a pack file.

Parameters:: location (PackChunkLocation) – Stored chunk location
Returns:: Chunk payload bytes
Return type:: bytes
Raises:: IntegrityError – Raised when the pack is missing or truncated.

Example:

>>> from pathlib import Path
>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     store = PackStore(Path(tmpdir))
...     result = store.write_pack("demo", [b"abc"])
...     store.read_chunk(result.chunks[0])
b'abc'

read_range(pack_id: str, offset: int, length: int) → bytes[source]

Read a byte range from a pack file.

Parameters:

pack_id (str) – Pack identifier without the file suffix
offset (int) – Absolute byte offset inside the pack file
length (int) – Number of bytes to read

Returns:

Requested byte range

Return type:

bytes

Raises:

IntegrityError – Raised when the pack header is invalid, the pack is missing, or the requested range exceeds the pack size.
ValueError – Raised when offset or length is negative.

Example:

>>> from pathlib import Path
>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     store = PackStore(Path(tmpdir))
...     result = store.write_pack("demo", [b"abcdef"])
...     store.read_range("demo", result.chunks[0].offset + 1, 3)
b'bcd'

write_pack(pack_id: str, chunks: Sequence[bytes]) → PackWriteResult[source]

Write an append-only pack file for ordered chunk bytes.

Parameters:

pack_id (str) – Pack identifier without the file suffix
chunks (Sequence[bytes]) – Ordered chunk payloads to append

Returns:

Pack metadata and chunk locations

Return type:

PackWriteResult

Raises:

IntegrityError – Raised when the target pack already exists.

Example:

>>> from pathlib import Path
>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     store = PackStore(Path(tmpdir))
...     result = store.write_pack("demo", [b"ab", b"cd"])
...     len(result.chunks)
2