"""Object transport and indexed-record extension contracts."""
from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import IO, Protocol, runtime_checkable
[docs]
@dataclass(frozen=True, slots=True)
class ObjectInfo:
size: int
generation: int | None = None
[docs]
@dataclass(frozen=True, slots=True)
class ObjectVersion:
data: bytes
generation: int | None = None
@runtime_checkable
class ObjectDownloader(Protocol):
"""Optional store capability: stream an object without buffering it whole.
A Protocol rather than an abstract method, so that adding it cannot break a
third-party ``ObjectStore``. Staging falls back to ``read`` when a store
does not offer it, at the cost of holding one shard in memory.
"""
def download_to_fileobj(
self, uri: str, stream: IO[bytes], *, generation: int | None = None
) -> None: ...
[docs]
class ObjectStore(ABC):
"""Immutable-object transport with version-aware control updates."""
#: Whether this store can modify an object in place without losing a
#: concurrent writer's update.
#:
#: Almost everything in the format is written once, so a store that cannot
#: do this is still fully usable for publishing datasets. The exception is
#: the annotation catalog, which is swung from one revision to the next
#: under a compare-and-swap; a store answering False refuses that, and
#: callers check here before doing any work rather than failing at the swap.
#:
#: A plain class attribute rather than an abstract member, so that every
#: ObjectStore written against an earlier release keeps working unchanged.
supports_compare_and_swap: bool = True
[docs]
@abstractmethod
def read(self, uri: str, *, generation: int | None = None) -> bytes: ...
[docs]
@abstractmethod
def read_version(self, uri: str) -> ObjectVersion: ...
[docs]
@abstractmethod
def info(self, uri: str, *, generation: int | None = None) -> ObjectInfo: ...
[docs]
@abstractmethod
def create(self, uri: str, data: bytes, *, content_type: str) -> int | None: ...
[docs]
@abstractmethod
def create_file(self, uri: str, path: Path, *, content_type: str) -> int | None: ...
[docs]
@abstractmethod
def compare_and_swap(
self, uri: str, data: bytes, *, expected_generation: int | None, content_type: str
) -> int | None: ...
[docs]
@abstractmethod
def generation(self, uri: str) -> int | None: ...
[docs]
class ShardStore(ABC):
"""Indexed byte-record container independent of logical codecs."""
[docs]
@abstractmethod
def write(self, path: Path, records: Iterable[bytes], *, options: str) -> int: ...
[docs]
@abstractmethod
def read(
self,
uri: str,
index: int,
*,
generation: int | None = None,
options: str,
file_reader_buffer_size: int | None = None,
) -> bytes: ...
[docs]
def read_many(
self,
uri: str,
indices: Sequence[int],
*,
generation: int | None = None,
options: str,
file_reader_buffer_size: int | None = None,
) -> tuple[bytes, ...]:
"""Read several entries while preserving the requested index order.
Implementations may override this method to reuse one underlying shard
reader. The default preserves compatibility for custom shard stores.
"""
return tuple(
self.read(
uri,
index,
generation=generation,
options=options,
file_reader_buffer_size=file_reader_buffer_size,
)
for index in indices
)
[docs]
@abstractmethod
def count(
self,
uri: str,
*,
generation: int | None = None,
options: str,
file_reader_buffer_size: int | None = None,
) -> int: ...