buildgrid.server.cas.storage.storage_abc module

StorageABC

The abstract base class for storage providers.

buildgrid.server.cas.storage.storage_abc.create_write_session(digest: Digest) → IO[bytes]

Return a file-like object to which a blob’s contents could be written.

For large files, to avoid excess memory usage, upload to temporary file. For small files we can work in memory for performance.

class buildgrid.server.cas.storage.storage_abc.StorageABC

Bases: ABC

TYPE: str

start() → None

stop() → None

abstract has_blob(digest: Digest) → bool

Return True if the blob with the given instance/digest exists. Unlike missing_blobs this method should NOT refresh the TTL of the underlying blob, if tracked.

abstract get_blob(digest: Digest) → IO[bytes] | None

Return a file-like object containing the blob. Most implementations will read the entire file into memory and return a BytesIO object. Eventually this should be corrected to handle files which cannot fit into memory.

The file-like object must be readable and seekable.

If the blob isn’t present in storage, return None.

stream_read_blob(digest: Digest, chunk_size: int, offset: int = 0, limit: int = 0) → Iterator[bytes]

Return a generator that yields the blob in chunks.

If the blob isn’t present in storage, it throws NotFound.

stream_write_blob(digest: Digest, chunks: Iterator[bytes]) → None

Given a stream of chunks, write it to the storage.

abstract delete_blob(digest: Digest) → None

Delete the blob from storage if it’s present.

abstract commit_write(digest: Digest, write_session: IO[bytes]) → None

Store the contents for a digest.

The storage object is not responsible for verifying that the data written to the write_session actually matches the digest. The caller must do that.

abstract bulk_delete(digests: list[Digest]) → list[str]

Delete a list of blobs from storage.

abstract missing_blobs(digests: list[Digest]) → list[Digest]

Return a container containing the blobs not present in CAS.

abstract bulk_update_blobs(blobs: list[tuple[Digest, bytes]]) → list[Status]

Given a container of (digest, value) tuples, add all the blobs to CAS. Return a list of Status objects corresponding to the result of uploading each of the blobs.

The storage object is not responsible for verifying that the data for each blob actually matches the digest. The caller must do that.

abstract bulk_read_blobs(digests: list[Digest]) → dict[str, bytes]

Given an iterable container of digests, return a {hash: file-like object} dictionary corresponding to the blobs represented by the input digests.

Each file-like object must be readable and seekable.

support_chunk_mappings() → bool

Return True if this storage can record and resolve spliced-blob chunk mappings.

Gates whether a SplitSplicer is constructed for the read fallback that reconstructs a spliced blob from its chunks. Only splice-capable storages (and wrappers delegating to them) return True; the default is False. Distinct from the Split/Splice RPC gate.

get_chunk_digests(blob_digest: Digest) → list[Digest] | None

Return the ordered chunk digests composing a spliced blob, or None if the blob is not spliced (or this storage doesn’t support splice).

record_spliced_blob(blob_digest: Digest, chunk_digests: Sequence[Digest]) → None

Persist an already-verified mapping of chunk_digests -> blob_digest.

The storage-specific counterpart driven by SplitSplicer; only splice-capable storages implement it.

bulk_get_chunk_digests(blob_digests: Sequence[Digest]) → dict[str, list[Digest]]

Return the chunk mappings for the spliced blobs among blob_digests, keyed by blob hash. Non-spliced (or unknown) digests are omitted.

Batched counterpart to get_chunk_digests(), letting a bulk read resolve every mapping in one query instead of one per digest. Splice-capable storages override it; the default falls back to per-digest lookups.

put_message(message: MessageType) → Digest

Store the given Protobuf message in CAS, returning its digest.

get_message(digest: Digest, message_type: type[M]) → M | None

Retrieve the Protobuf message with the given digest and type from CAS. If the blob is not present, returns None.

get_tree(root_digest: Digest, raise_on_missing_subdir: bool = False) → Iterator[Directory]