herostools.actor.archiver ========================= .. py:module:: herostools.actor.archiver Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/herostools/actor/archiver/array/index /autoapi/herostools/actor/archiver/artifact_storage/index /autoapi/herostools/actor/archiver/base/index /autoapi/herostools/actor/archiver/json_archiver/index /autoapi/herostools/actor/archiver/postgres/index /autoapi/herostools/actor/archiver/zarr_archiver/index Exceptions ---------- .. autoapisummary:: herostools.actor.archiver.ArtifactReadError Classes ------- .. autoapisummary:: herostools.actor.archiver.ArrayArchiver herostools.actor.archiver.ArtifactStorage herostools.actor.archiver.InMemoryArtifactStorage herostools.actor.archiver.S3ArtifactStorage herostools.actor.archiver.HERODataArchiver herostools.actor.archiver.JsonArchiver herostools.actor.archiver.PostgresArchiver herostools.actor.archiver.PostgresDatabase herostools.actor.archiver.PostgresRecordChangeListener herostools.actor.archiver.PostgresRecordStore herostools.actor.archiver.ZarrArchiver Package Contents ---------------- .. py:class:: ArrayArchiver(save_template: str, split_data_array: bool = False, *args, **kwargs) Bases: :py:obj:`herostools.actor.archiver.base.HERODataArchiver` HERODataArchiver that saves numpy arrays as ``.npy`` files. :param object_selector: Zenoh object selector for the devices to subscribe to. :param event_name: Name of the event. :param save_template: Jinja2 template used to generate the output file path from the payload metadata. :param split_data_array: When ``True``, each row of a 2-D array is saved as a separate file. Use ``{{ _split_index }}`` in ``save_template`` to embed the row index in the filename. :param default_metadata: Default metadata available to the filename template. :param max_retries: Number of times to retry a failed store before dropping the item. .. rubric:: Example BOSS json configuration:: { "_id": "my-camera-capturer", "classname": "herostools.archiver.ArrayArchiver", "arguments": { "object_selector": "my-camera", "event_name": "acquisition_data", "default_metadata": { "file_path": "/mnt/mystorage/images" }, "save_template": "{{ file_path }}/testimg-{{ '%04d' % ( frame / 2 ) |round(0, 'floor') }}-{{ frame % 2 }}.npy" } } Generates paths such as:: /mnt/mystorage/images/testimg-0000-0.npy /mnt/mystorage/images/testimg-0000-1.npy .. py:attribute:: name_template .. py:attribute:: split_data_array :value: False .. py:method:: _store(source_name: str, payload: numpy.typing.NDArray[Any], metadata: dict) -> None Save the payload as a numpy ``.npy`` file. :param source_name: Name of the event source (the HERO). :param payload: Data to save as a numpy array. :param metadata: Incoming metadata merged with ``default_metadata``. .. 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. .. py:class:: HERODataArchiver(object_selector: str, event_name: str, default_metadata: dict | None = None, max_retries: int = 5, *args, **kwargs) Bases: :py:obj:`heros.EventObserver` Base EventObserver that subscribes to a HERO event and archives its payload. The event payload may be a ``(data, metadata)`` tuple or a plain dict. Subclasses implement :meth:`_store`. :param object_selector: Zenoh object selector for the devices to subscribe to. :param event_name: Name of the event. :param default_metadata: Default metadata merged with every incoming payload's metadata. :param max_retries: Number of times to retry a failed store before dropping the item. .. py:attribute:: metadata .. py:attribute:: max_retries :value: 5 .. py:attribute:: _payload_queue .. py:attribute:: _stop_event .. py:attribute:: _worker_thread .. py:method:: _process_queue() -> None Background worker that consumes the payload queue. .. py:method:: is_queue_drained() -> bool Return True when all queued items have been processed. :returns: True if every item put on the queue has had task_done() called. .. py:method:: _stop() Stop the background thread gracefully. .. py:method:: _teardown() -> None Called by boss on shutdown. .. py:method:: feed(source_name: str, data: Iterable, retry_count: int = 0) -> None Callback registered with the source event. ``data`` may be a ``(payload, metadata)`` tuple or a plain dict. When a plain dict is received it is used as both payload and metadata so downstream ``_store`` implementations can still locate identifier keys regardless of which field they are in. :param source_name: Name of the event source (the HERO). :param data: Either a ``(payload, metadata)`` tuple or a plain dict payload. :param retry_count: Number of times this item has already been retried. .. py:method:: _store(source_name: str, payload: Any, metadata: dict) -> None :abstractmethod: Store the payload. Must be implemented by subclasses. :param source_name: Name of the event source (the HERO). :param payload: The actual data. :param metadata: Incoming metadata merged with ``default_metadata``. .. py:class:: JsonArchiver(save_template: str, merge_metadata: bool = False, *args, **kwargs) Bases: :py:obj:`herostools.actor.archiver.base.HERODataArchiver` HERODataArchiver that saves dict payloads as ``.json`` files. :param object_selector: Zenoh object selector for the devices to subscribe to. :param event_name: Name of the event. :param save_template: Jinja2 template used to generate the output file path from the payload metadata. :param merge_metadata: When ``True``, the metadata dict is embedded in the saved JSON under the key ``metadata``. :param default_metadata: Default metadata available to the filename template. :param max_retries: Number of times to retry a failed store before dropping the item. .. py:attribute:: name_template .. py:attribute:: merge_metadata :value: False .. py:method:: _store(source_name: str, payload: dict, metadata: dict) -> None Save the payload as a ``.json`` file. :param source_name: Name of the event source (the HERO). :param payload: Data to save as a JSON-serialisable dict. :param metadata: Incoming metadata merged with ``default_metadata``. .. py:class:: PostgresArchiver(db_url: str, identifier_key: str = 'identifier', allow_purge: bool = False, artifact_storage_kwargs: dict | None = _USE_DEFAULT_S3, array_key_template: str = '{{ source_name }}', use_single_bucket: bool = False, retention_days: int = 0, artifact_storage: herostools.actor.archiver.artifact_storage.ArtifactStorage | None = None, *args, **kwargs) Bases: :py:obj:`herostools.actor.archiver.base.HERODataArchiver`, :py:obj:`PostgresRecordStore` HEROS event adapter for :class:`PostgresRecordStore`. Subscribes to incoming data events, queues and retries writes, and publishes ``record_changed`` notifications. Use :class:`PostgresRecordStore` for local current-state queries without HEROS or notification connections. .. py:attribute:: _change_listener .. py:attribute:: _listener_thread .. py:method:: record_changed(identifiers: list[str]) -> list[str] Publish changed record identifiers to HEROS subscribers. :param identifiers: Deduplicated identifiers from PostgreSQL notifications. :returns: The published identifiers. .. py:method:: _store(source_name: str, payload: Any, metadata: dict) -> None Store a queued HEROS payload through the record store. :param source_name: Name of the event source. :param payload: The data to store. :param metadata: Metadata merged into the payload. .. py:method:: _listen_loop() -> None Publish PostgreSQL change notifications through the HEROS event. .. py:method:: _process_queue() -> None Drain the queue, then close this thread's DB connection on exit. .. py:method:: _teardown() -> None Stop worker and listener threads; worker closes its own DB connection. .. py:class:: PostgresDatabase(db_url: str, identifier_key: str = 'identifier', allow_purge: bool = False, retention_days: int = 0) Pure SQL layer for the archiver schema — stores and retrieves clean JSON. Can be used independently of HEROS to insert, query, and delete records. Payloads must be clean JSON dicts (no numpy arrays). Artifact references stored in ``data_json`` are returned as plain dicts by :meth:`get` without resolution; use :class:`PostgresArchiver` for the full artifact pipeline. Each identifier maps to exactly one row in ``records``. Successive payloads for the same identifier are shallow-merged (``||``) at write time; later events win on key collision. :param db_url: psycopg connection string. :param identifier_key: Key used to look up the record identifier in the payload. :param allow_purge: Enable the :meth:`purge` method (disabled by default). :param retention_days: Default retention window in days used by :meth:`_find_expired`. ``0`` means keep forever. .. py:attribute:: _db_url .. py:attribute:: _local .. py:attribute:: _identifier_key :value: 'identifier' .. py:attribute:: _allow_purge :value: False .. py:attribute:: _retention_days :value: 0 .. py:method:: _get_conn() -> psycopg.Connection Return a per-thread connection, opening one if necessary. psycopg connections are not thread-safe; each thread owns its own connection. .. py:method:: close() -> None Close the calling thread's database connection. .. py:method:: _store(payload: dict) -> str | None Upsert a clean JSON payload into records, merging into any existing row. The identifier is extracted from the payload using :attr:`_identifier_key`. Returns ``None`` and logs an error when the identifier key is missing. :param payload: Clean JSON dict to store. Must not contain numpy arrays. :returns: The identifier string, or ``None`` if the identifier key is absent. .. py:method:: get_ids(tag: str, values: list[Any]) -> list[str] Return identifiers matching the given tag/key in either source. Searches ``identifier_groups`` (post-hoc tags) and ``records.data_json`` (payload fields). Results are deduplicated and sorted. :param tag: Tag key (``identifier_groups``) or top-level ``data_json`` key. :param values: Accepted values (cast to str for comparison). :returns: Sorted, deduplicated list of matching identifiers. .. py:method:: get(ids: list[str]) -> dict[str, dict] Fetch the merged record for each identifier, including post-hoc tags. Tags stored via :meth:`tag_ids` are merged into the returned dict alongside ``data_json`` fields. Tags overwrite ``data_json`` values on key collision. Artifact references remain as plain dicts in the result. :param ids: Identifiers to fetch. :returns: Dict mapping identifier to the merged record dict. Identifiers with no record are absent from the result. .. py:method:: tag_ids(ids: list[str], tag: str, value: str) -> None Attach a tag to a list of identifiers, overwriting any existing value for that tag. Inserts ``(identifier, tag, value)`` into ``identifier_groups`` for each identifier. If the ``(identifier, tag)`` pair already exists the value is updated in place. :param ids: Identifiers to tag. Each must already exist in ``records``. :param tag: Tag key. :param value: Tag value to assign. :raises psycopg.errors.ForeignKeyViolation: If any identifier is not present in ``records``. .. py:method:: purge(ids: list[str]) -> None Delete DB rows for the given identifiers from records and identifier_groups. Requires ``allow_purge=True`` at instantiation time. Does not touch artifact storage; use :meth:`PostgresArchiver.purge` for that. :param ids: Identifiers to delete. :raises PermissionError: If ``allow_purge`` was not set to True. .. py:method:: _find_expired(days: int) -> list[str] Return identifiers older than ``days`` days, excluding saved ones. Records tagged with ``_save`` set to a truthy value (``'1'``, ``'true'``, ``'True'``) are excluded from the result. :param days: Retention window in days. :returns: List of expired identifier strings. .. py:class:: PostgresRecordChangeListener(db_url: str) Yield batches of changed record identifiers from PostgreSQL notifications. :param db_url: Psycopg connection string. .. py:attribute:: _db_url .. py:method:: iter_changes(stop_event: threading.Event) -> collections.abc.Iterator[list[str]] Yield deduplicated notification batches until stopped. :param stop_event: Event that stops the notification loop. :Yields: Lists of changed record identifiers. .. py:class:: PostgresRecordStore(db_url: str, identifier_key: str = 'identifier', allow_purge: bool = False, artifact_storage_kwargs: dict | None = _USE_DEFAULT_S3, array_key_template: str = '{{ source_name }}', use_single_bucket: bool = False, retention_days: int = 0, artifact_storage: herostools.actor.archiver.artifact_storage.ArtifactStorage | None = None) Bases: :py:obj:`PostgresDatabase` HEROS-independent PostgreSQL record store with artifact handling. Owns the full artifact pipeline: numpy array detection, artifact storage writes and reads, per-bucket resolution, and artifact deletion on purge. :class:`PostgresDatabase` handles only SQL; this class adds the artifact layer. It does not subscribe to HEROS events or PostgreSQL notifications. Each identifier maps to one row in ``records``. Successive payloads are shallow-merged at write time (``||``); later events win on key collision. Numpy arrays are replaced by artifact_ref dicts in ``data_json`` and stored in the configured artifact storage. Each ref embeds the bucket it was written to, enabling transparent multi-bucket resolution and deletion. :param db_url: psycopg connection string. :param identifier_key: Key used to look up the record identifier. :param allow_purge: Enable the :meth:`purge` method. :param artifact_storage_kwargs: Passed as ``S3ArtifactStorage(**kwargs)``. Pass ``None`` to use ``InMemoryArtifactStorage`` (no persistence, useful for dev/test). Omit to use local RustFS defaults. :param array_key_template: Jinja2 template rendered against per-event metadata to form the dict key for bare numpy-array payloads. ``source_name`` is always available in the template context alongside all metadata fields. Defaults to ``"{{ source_name }}"`` which preserves existing behaviour. :param use_single_bucket: Store new artifacts in the configured static bucket instead of UTC daily buckets. Defaults to False. :param retention_days: Default retention window in days used by :meth:`cleanup`. ``0`` (default) means keep forever. :param artifact_storage: Explicit storage instance. Mutually exclusive with ``artifact_storage_kwargs``. .. py:attribute:: _artifact_storage .. py:attribute:: _array_key_template .. py:attribute:: _use_single_bucket :value: False .. py:method:: store(source_name: str, payload: Any, metadata: dict) -> None Store one payload after extracting any numpy-array artifacts. A bare numpy array payload is wrapped as ``{template_key: array}`` first. ``metadata`` is shallow-merged into the payload (payload wins on key collision). Arrays are replaced by artifact_ref dicts with the bucket embedded; artifacts are uploaded to storage before the DB write. :param source_name: Name of the event source. :param payload: The actual data (dict or numpy array). :param metadata: Incoming metadata merged with ``default_metadata``. .. py:method:: _store(source_name: str, payload: Any, metadata: dict) -> None Store one payload through the synchronous record-store API. :param source_name: Name of the event source. :param payload: The data to store. :param metadata: Metadata merged into the payload. .. py:method:: get(ids: list[str], resolve_artifacts: bool = True, size_limit_bytes: int | None = None) -> dict[str, dict] Fetch merged records and optionally resolve artifact_refs to numpy arrays. Tags stored via :meth:`tag_ids` are merged into the returned dict alongside ``data_json`` fields. Tags overwrite ``data_json`` values on key collision (post-hoc annotations take priority). Artifact_refs are resolved per-bucket: each ref's stored ``bucket`` field determines which bucket the array is fetched from. Refs without a ``bucket`` field (legacy records) fall back to the storage ``default_bucket``. :param ids: Identifiers to fetch. :param resolve_artifacts: Replace artifact_ref dicts with numpy arrays. :param size_limit_bytes: When set, only artifacts whose stored ``size_bytes`` is at or below this threshold are fetched. Larger artifacts remain as artifact_ref dicts. Refs without ``size_bytes`` are always resolved. ``None`` resolves all regardless of size. :returns: Dict mapping identifier to the merged record dict. Identifiers with no record are absent from the result. .. py:method:: get_artifact(artifact_id: str, bucket: str | None = None) -> numpy.ndarray Retrieve one decoded artifact from the configured artifact storage. :param artifact_id: Identifier of the artifact to retrieve. :param bucket: Bucket override; uses the storage default when None. :returns: The decoded numpy array. .. py:method:: get_artifacts(locations: collections.abc.Sequence[tuple[str, str | None]]) -> collections.abc.Mapping[str, numpy.ndarray] Retrieve a bounded set of artifacts from the configured storage. :param locations: Artifact ID and optional bucket pairs. :returns: Mapping of artifact IDs to decoded numpy arrays. .. py:method:: purge(ids: list[str], remove_artifacts: bool = True) -> None Delete all data for the given identifiers, including stored artifacts. Artifacts are removed per-bucket using each ref's stored ``bucket`` field. Refs without a ``bucket`` field fall back to ``default_bucket``. DB rows are deleted after artifact removal. Requires ``allow_purge=True`` at instantiation time. :param ids: Identifiers to purge. :param remove_artifacts: Also remove referenced artifacts from storage. :raises PermissionError: If ``allow_purge`` was not set to True. .. py:method:: cleanup(retention_days: int | None = None, remove_artifacts: bool = True, dry_run: bool = True) -> list[str] Delete records older than the retention window, skipping saved ones. Requires ``allow_purge=True``. Records tagged with ``_save`` set to a truthy value (``'1'``, ``'true'``, ``'True'``) are never deleted. Returns the list of expired identifiers (deleted or would-be-deleted). :param retention_days: Purge records whose ``updated_at`` is older than this many days. ``None`` uses the instance default set at construction. ``0`` is a no-op (keep forever). :param remove_artifacts: Also remove referenced artifacts from storage. :param dry_run: When True (default), only return the expired identifiers without deleting anything. :returns: List of expired identifiers. :raises PermissionError: If ``allow_purge`` was not set to True. .. py:class:: ZarrArchiver(store_template: str, array_path_template: str, *args, **kwargs) Bases: :py:obj:`herostools.actor.archiver.base.HERODataArchiver` HERODataArchiver that stores numpy array payloads in a zarr store. The store root and the array key within it are both derived from Jinja2 templates rendered against the merged metadata. Opening the store in append mode ('a') means successive calls accumulate arrays in the same store directory. :param store_template: Jinja2 template rendered to the zarr store root path. :param array_path_template: Jinja2 template rendered to the array key within the store. :param object_selector: Zenoh object selector for the devices to subscribe to. :param event_name: Name of the event. :param default_metadata: Default metadata available to both templates. :param max_retries: Number of times to retry a failed store before dropping the item. .. py:attribute:: store_template .. py:attribute:: array_path_template .. py:method:: _store(source_name: str, payload: numpy.typing.NDArray[Any], metadata: dict) -> None Write the payload array into the zarr store. :param source_name: Name of the event source (the HERO). :param payload: Numpy array to store. :param metadata: Incoming metadata merged with ``default_metadata``.