herostools.actor.archiver.postgres
==================================

.. py:module:: herostools.actor.archiver.postgres


Attributes
----------

.. autoapisummary::

   herostools.actor.archiver.postgres._DROPPED
   herostools.actor.archiver.postgres._USE_DEFAULT_S3


Classes
-------

.. autoapisummary::

   herostools.actor.archiver.postgres.PostgresDatabase
   herostools.actor.archiver.postgres.PostgresRecordChangeListener
   herostools.actor.archiver.postgres.PostgresRecordStore
   herostools.actor.archiver.postgres.PostgresArchiver


Functions
---------

.. autoapisummary::

   herostools.actor.archiver.postgres._daily_artifact_bucket
   herostools.actor.archiver.postgres._substitute_artifacts
   herostools.actor.archiver.postgres._collect_artifact_refs
   herostools.actor.archiver.postgres._resolve_artifacts_batch


Module Contents
---------------

.. py:data:: _DROPPED

.. py:function:: _daily_artifact_bucket(default_bucket: str | None, now: datetime.datetime | None = None) -> str | None

   Return the UTC daily bucket derived from a storage default bucket.

   :param default_bucket: Static bucket used only for legacy-reference fallback.
   :param now: Optional timestamp, used by tests to make routing deterministic.

   :returns: Daily bucket name, or None for storage backends without buckets.


.. py:function:: _substitute_artifacts(obj: Any, bucket: str | None = None) -> tuple[Any, list[tuple[uuid.UUID, str, numpy.ndarray]]]

   Recursively replace numpy arrays with artifact refs; drop other non-JSON values.

   The artifact name for each array is its dot-separated key path within the
   original dict (e.g. "sensor.spectrum" for a nested key, "items.1" for a
   list element at index 1). Each artifact_ref dict includes intrinsic array
   metadata: ``size_bytes`` (element data only, excluding the npy header),
   ``shape``, and ``dtype``. ``bucket`` is the storage bucket name at write
   time (``None`` when no storage backend is configured).

   :param obj: Dict or list to walk.
   :param bucket: Bucket name to embed in every artifact_ref produced.

   :returns: Tuple of (cleaned_obj, artifacts) where artifacts is a list of
             (artifact_id, artifact_name, array) triples.


.. py:function:: _collect_artifact_refs(obj: Any) -> list[dict]

   Recursively collect all artifact_ref dicts.

   :param obj: Dict, list, or scalar to walk.

   :returns: List of artifact_ref dicts found.


.. py:function:: _resolve_artifacts_batch(obj: Any, lookup: dict[str, numpy.ndarray]) -> Any

   Recursively replace artifact_ref dicts using a pre-fetched id->array lookup.

   :param obj: Dict, list, or scalar to walk.
   :param lookup: Mapping of artifact_id strings to numpy arrays.

   :returns: Object with all artifact_ref dicts replaced by numpy arrays.


.. 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:data:: _USE_DEFAULT_S3

.. 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:: 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.



