herostools.actor.archiver.artifact_storage ========================================== .. py:module:: herostools.actor.archiver.artifact_storage Attributes ---------- .. autoapisummary:: herostools.actor.archiver.artifact_storage.DEFAULT_MAX_ARTIFACTS_PER_BATCH Exceptions ---------- .. autoapisummary:: herostools.actor.archiver.artifact_storage.ArtifactReadError Classes ------- .. autoapisummary:: herostools.actor.archiver.artifact_storage.ArtifactStorage herostools.actor.archiver.artifact_storage.InMemoryArtifactStorage herostools.actor.archiver.artifact_storage.S3ArtifactStorage Module Contents --------------- .. py:data:: DEFAULT_MAX_ARTIFACTS_PER_BATCH :value: 100 .. py:exception:: ArtifactReadError(artifact_id: str, bucket: str | None) Bases: :py:obj:`RuntimeError` Raised when an artifact cannot be read from its configured storage location. :param artifact_id: Identifier of the artifact that could not be read. :param bucket: Effective bucket used for the read. .. py:attribute:: artifact_id .. py:attribute:: bucket .. py:class:: ArtifactStorage(max_artifacts_per_batch: int = DEFAULT_MAX_ARTIFACTS_PER_BATCH) Bases: :py:obj:`abc.ABC` Abstract base for artifact key-value stores. Implementations must support put/get/remove by string artifact_id and optionally list all stored IDs. All mutating methods accept an optional ``bucket`` override; when omitted they use the backend's default bucket. .. py:attribute:: _max_artifacts_per_batch :value: 100 .. py:property:: default_bucket :type: str | None Default bucket name for this storage backend, or None if not applicable. .. py:method:: put(artifact_id: str, data: numpy.ndarray, bucket: str | None = None) -> str | None :abstractmethod: Store a numpy array under the given artifact_id. :param artifact_id: Unique key (UUID string). :param data: Array to store. :param bucket: Bucket override; uses backend default when None. :returns: The effective bucket name used for storage, or None for backends without a bucket concept (e.g. in-memory). .. py:method:: get(artifact_id: str, bucket: str | None = None) -> numpy.ndarray :abstractmethod: Retrieve a numpy array by artifact_id. :param artifact_id: Key to look up. :param bucket: Bucket override; uses backend default when None. :returns: The stored numpy array. .. py:method:: remove(artifact_id: str, bucket: str | None = None) -> None :abstractmethod: Delete the artifact with the given artifact_id. :param artifact_id: Key to delete. :param bucket: Bucket override; uses backend default when None. .. py:method:: list(bucket: str | None = None) -> list[str] :abstractmethod: Return all artifact_ids currently stored. :param bucket: Bucket override; uses backend default when None. :returns: List of artifact_id strings. .. py:method:: get_many(artifact_ids: list[str], bucket: str | None = None) -> list[numpy.ndarray] Retrieve multiple arrays by artifact_id, in the same order as the input. :param artifact_ids: Keys to look up. :param bucket: Bucket override forwarded to each :meth:`get` call. :returns: List of numpy arrays in the same order as artifact_ids. .. py:method:: get_artifact(artifact_id: str, bucket: str | None = None) -> numpy.ndarray Retrieve one decoded artifact by its ID and optional bucket. :param artifact_id: Identifier of the artifact to retrieve. :param bucket: Bucket override; uses :attr:`default_bucket` when None. :returns: The decoded numpy array. :raises ArtifactReadError: If the artifact cannot be read. .. py:method:: get_artifacts(locations: collections.abc.Sequence[tuple[str, str | None]]) -> collections.abc.Mapping[str, numpy.ndarray] Retrieve a bounded set of artifacts addressed by ID and bucket. The returned mapping is keyed only by artifact ID. An ID may therefore occur at one effective bucket only within a request. Identical locations are read once and represented once in the result. :param locations: Artifact ID and optional bucket pairs. :returns: Mapping of artifact IDs to decoded numpy arrays. :raises ValueError: If the request exceeds the batch limit or one artifact ID is requested from multiple effective buckets. :raises ArtifactReadError: If an artifact cannot be read. .. py:method:: ensure_bucket(bucket: str | None) -> None Ensure a bucket is available for writes when the backend requires it. :param bucket: Bucket to ensure, or None for backends without buckets. .. py:method:: put_many(artifact_ids: list[str], arrays: list[numpy.ndarray], bucket: str | None = None) -> None Store multiple arrays under their respective artifact_ids. :param artifact_ids: Unique keys. :param arrays: Arrays to store, paired with artifact_ids by position. :param bucket: Bucket override forwarded to each :meth:`put` call. .. py:method:: remove_many(artifact_ids: list[str], bucket: str | None = None) -> None Delete multiple artifacts by artifact_id. :param artifact_ids: Keys to delete. :param bucket: Bucket override forwarded to each :meth:`remove` call. .. py:class:: InMemoryArtifactStorage(max_artifacts_per_batch: int = DEFAULT_MAX_ARTIFACTS_PER_BATCH) Bases: :py:obj:`ArtifactStorage` Dict-backed in-memory ArtifactStorage — useful for testing and dry runs. The ``bucket`` parameter is accepted on all methods for interface compatibility but is ignored; all artifacts share a single in-memory dict regardless of bucket. .. py:attribute:: _store :type: dict[str, numpy.ndarray] .. py:method:: put(artifact_id: str, data: numpy.ndarray, bucket: str | None = None) -> str | None Store array in memory. :param artifact_id: Unique key. :param data: Array to store. :param bucket: Ignored. :returns: None (no bucket concept for in-memory storage). .. py:method:: get(artifact_id: str, bucket: str | None = None) -> numpy.ndarray Retrieve array from memory. :param artifact_id: Key to look up. :param bucket: Ignored. :returns: The stored numpy array. .. py:method:: remove(artifact_id: str, bucket: str | None = None) -> None Remove array from memory. :param artifact_id: Key to delete. :param bucket: Ignored. .. py:method:: list(bucket: str | None = None) -> list[str] Return all stored keys. :param bucket: Ignored. :returns: List of artifact_id strings. .. py:class:: S3ArtifactStorage(endpoint_url: str | None = 'http://localhost:9000', bucket: str = 'artifacts', access_key: str = 'heros', secret_key: str = 'heros', max_workers: int = 10, max_artifacts_per_batch: int = DEFAULT_MAX_ARTIFACTS_PER_BATCH, create_buckets: bool = True) Bases: :py:obj:`ArtifactStorage` ArtifactStorage backed by any S3-compatible object store. Numpy arrays are serialised via np.save/np.load into BytesIO and stored as S3 objects keyed by artifact_id (UUID string). The bucket is created on first use if it does not already exist. Defaults point to the RustFS instance defined in docker-compose.yml (endpoint ``http://localhost:9000``, credentials ``rustfs``/``rustfsdev``). To use AWS S3, pass ``endpoint_url=None`` and supply real IAM credentials:: S3ArtifactStorage( endpoint_url=None, bucket="my-bucket", access_key="MY_ACCESS_KEY", secret_key="MY_SECRET_KEY", ) :param endpoint_url: S3 API URL for self-hosted backends, e.g. ``"http://localhost:9000"`` for local RustFS. Pass ``None`` to use the default AWS S3 endpoint. :param bucket: Default bucket name. :param access_key: Access key id (``RUSTFS_ACCESS_KEY`` for RustFS, AWS access key id for AWS S3). :param secret_key: Secret access key (``RUSTFS_SECRET_KEY`` for RustFS, AWS secret access key for AWS S3). :param create_buckets: Create missing buckets on construction and before writes. Set False for read-only clients. .. py:attribute:: _bucket :value: 'artifacts' .. py:attribute:: _max_workers :value: 10 .. py:attribute:: _create_buckets :value: True .. py:attribute:: _known_buckets :type: set[str] .. py:attribute:: _bucket_lock .. py:attribute:: _client .. py:property:: default_bucket :type: str Default bucket name configured at construction time. .. py:method:: ensure_bucket(bucket: str | None) -> None Ensure an S3 bucket exists, tolerating concurrent creation. :param bucket: Bucket to ensure. None uses :attr:`default_bucket`. :raises ClientError: If the bucket cannot be inspected or created. .. py:method:: put(artifact_id: str, data: numpy.ndarray, bucket: str | None = None) -> str Serialise array to npy bytes and upload as an S3 object. :param artifact_id: S3 object key. :param data: Array to store. :param bucket: Bucket override; uses :attr:`default_bucket` when None. :returns: The effective bucket name used for storage. .. py:method:: get(artifact_id: str, bucket: str | None = None) -> numpy.ndarray Download and deserialise an S3 object to a numpy array. :param artifact_id: S3 object key. :param bucket: Bucket override; uses :attr:`default_bucket` when None. :returns: The stored numpy array. .. py:method:: remove(artifact_id: str, bucket: str | None = None) -> None Delete an S3 object. :param artifact_id: S3 object key to delete. :param bucket: Bucket override; uses :attr:`default_bucket` when None. .. py:method:: list(bucket: str | None = None) -> list[str] Return all object keys in the bucket via paginated listing. :param bucket: Bucket override; uses :attr:`default_bucket` when None. :returns: List of artifact_id strings. .. py:method:: get_many(artifact_ids: list[str], bucket: str | None = None) -> list[numpy.ndarray] Download and deserialise multiple S3 objects in parallel. :param artifact_ids: S3 object keys to fetch. :param bucket: Bucket override forwarded to each :meth:`get` call. :returns: List of numpy arrays in the same order as artifact_ids. .. py:method:: put_many(artifact_ids: list[str], arrays: list[numpy.ndarray], bucket: str | None = None) -> None Serialise and upload multiple arrays to S3 in parallel. :param artifact_ids: S3 object keys. :param arrays: Arrays to store, paired with artifact_ids by position. :param bucket: Bucket override forwarded to each :meth:`put` call. .. py:method:: remove_many(artifact_ids: list[str], bucket: str | None = None) -> None Delete multiple S3 objects using the native batch delete API. Sends at most 1000 keys per request as required by the S3 API. :param artifact_ids: S3 object keys to delete. :param bucket: Bucket override; uses :attr:`default_bucket` when None.