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.