herostools.actor.archiver.postgres

Attributes

_DROPPED
_USE_DEFAULT_S3

Classes

PostgresDatabase	Pure SQL layer for the archiver schema — stores and retrieves clean JSON.
PostgresRecordChangeListener	Yield batches of changed record identifiers from PostgreSQL notifications.
PostgresRecordStore	HEROS-independent PostgreSQL record store with artifact handling.
PostgresArchiver	HEROS event adapter for PostgresRecordStore.

Functions

_daily_artifact_bucket(→ str | None)	Return the UTC daily bucket derived from a storage default bucket.
_substitute_artifacts(→ tuple[Any, ...)	Recursively replace numpy arrays with artifact refs; drop other non-JSON values.
_collect_artifact_refs(→ list[dict])	Recursively collect all artifact_ref dicts.
_resolve_artifacts_batch(→ Any)	Recursively replace artifact_ref dicts using a pre-fetched id->array lookup.

Module Contents

herostools.actor.archiver.postgres._DROPPED

herostools.actor.archiver.postgres._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.

Parameters:

• default_bucket – Static bucket used only for legacy-reference fallback.

• now – Optional timestamp, used by tests to make routing deterministic.

Returns:

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

herostools.actor.archiver.postgres._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).

Parameters:

• obj – Dict or list to walk.

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

herostools.actor.archiver.postgres._collect_artifact_refs(obj: Any) → list[dict]

Recursively collect all artifact_ref dicts.

Parameters:

obj – Dict, list, or scalar to walk.

Returns:

List of artifact_ref dicts found.

herostools.actor.archiver.postgres._resolve_artifacts_batch(obj: Any, lookup: dict[str, numpy.ndarray]) → Any

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

Parameters:

• obj – Dict, list, or scalar to walk.

• lookup – Mapping of artifact_id strings to numpy arrays.

Returns:

Object with all artifact_ref dicts replaced by numpy arrays.

class herostools.actor.archiver.postgres.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 get() without resolution; use 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.

Parameters:

• db_url – psycopg connection string.

• identifier_key – Key used to look up the record identifier in the payload.

• allow_purge – Enable the purge() method (disabled by default).

• retention_days – Default retention window in days used by _find_expired(). 0 means keep forever.

_db_url

_local

_identifier_key = 'identifier'

_allow_purge = False

_retention_days = 0

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

close() → None

Close the calling thread’s database connection.

_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 _identifier_key. Returns None and logs an error when the identifier key is missing.

Parameters:

payload – Clean JSON dict to store. Must not contain numpy arrays.

Returns:

The identifier string, or None if the identifier key is absent.

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.

Parameters:

• tag – Tag key (identifier_groups) or top-level data_json key.

• values – Accepted values (cast to str for comparison).

Returns:

Sorted, deduplicated list of matching identifiers.

get(ids: list[str]) → dict[str, dict]

Fetch the merged record for each identifier, including post-hoc tags.

Tags stored via 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.

Parameters:

ids – Identifiers to fetch.

Returns:

Dict mapping identifier to the merged record dict. Identifiers with no record are absent from the result.

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.

Parameters:

• ids – Identifiers to tag. Each must already exist in records.

• tag – Tag key.

• value – Tag value to assign.

Raises:

psycopg.errors.ForeignKeyViolation – If any identifier is not present in records.

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 PostgresArchiver.purge() for that.

Parameters:

ids – Identifiers to delete.

Raises:

PermissionError – If allow_purge was not set to True.

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

Parameters:

days – Retention window in days.

Returns:

List of expired identifier strings.

class herostools.actor.archiver.postgres.PostgresRecordChangeListener(db_url: str)

Yield batches of changed record identifiers from PostgreSQL notifications.

Parameters:

db_url – Psycopg connection string.

_db_url

iter_changes(stop_event: threading.Event) → collections.abc.Iterator[list[str]]

Yield deduplicated notification batches until stopped.

Parameters:

stop_event – Event that stops the notification loop.

Yields:

Lists of changed record identifiers.

herostools.actor.archiver.postgres._USE_DEFAULT_S3

class herostools.actor.archiver.postgres.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: 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. 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.

Parameters:

• db_url – psycopg connection string.

• identifier_key – Key used to look up the record identifier.

• allow_purge – Enable the purge() method.

• artifact_storage_kwargs – Passed as S3ArtifactStorage(**kwargs). Pass None to use InMemoryArtifactStorage (no persistence, useful for dev/test). Omit to use local RustFS defaults.

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

• use_single_bucket – Store new artifacts in the configured static bucket instead of UTC daily buckets. Defaults to False.

• retention_days – Default retention window in days used by cleanup(). 0 (default) means keep forever.

• artifact_storage – Explicit storage instance. Mutually exclusive with artifact_storage_kwargs.

_artifact_storage

_array_key_template

_use_single_bucket = False

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.

Parameters:

• source_name – Name of the event source.

• payload – The actual data (dict or numpy array).

• metadata – Incoming metadata merged with default_metadata.

_store(source_name: str, payload: Any, metadata: dict) → None

Store one payload through the synchronous record-store API.

Parameters:

• source_name – Name of the event source.

• payload – The data to store.

• metadata – Metadata merged into the payload.

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

Parameters:

• ids – Identifiers to fetch.

• resolve_artifacts – Replace artifact_ref dicts with numpy arrays.

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

get_artifact(artifact_id: str, bucket: str | None = None) → numpy.ndarray

Retrieve one decoded artifact from the configured artifact storage.

Parameters:

• artifact_id – Identifier of the artifact to retrieve.

• bucket – Bucket override; uses the storage default when None.

Returns:

The decoded numpy array.

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.

Parameters:

locations – Artifact ID and optional bucket pairs.

Returns:

Mapping of artifact IDs to decoded numpy arrays.

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.

Parameters:

• ids – Identifiers to purge.

• remove_artifacts – Also remove referenced artifacts from storage.

Raises:

PermissionError – If allow_purge was not set to True.

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

Parameters:

• 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).

• remove_artifacts – Also remove referenced artifacts from storage.

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

class herostools.actor.archiver.postgres.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: herostools.actor.archiver.base.HERODataArchiver, PostgresRecordStore

HEROS event adapter for PostgresRecordStore.

Subscribes to incoming data events, queues and retries writes, and publishes record_changed notifications. Use PostgresRecordStore for local current-state queries without HEROS or notification connections.

_change_listener

_listener_thread

record_changed(identifiers: list[str]) → list[str]

Publish changed record identifiers to HEROS subscribers.

Parameters:

identifiers – Deduplicated identifiers from PostgreSQL notifications.

Returns:

The published identifiers.

_store(source_name: str, payload: Any, metadata: dict) → None

Store a queued HEROS payload through the record store.

Parameters:

• source_name – Name of the event source.

• payload – The data to store.

• metadata – Metadata merged into the payload.

_listen_loop() → None

Publish PostgreSQL change notifications through the HEROS event.

_process_queue() → None

Drain the queue, then close this thread’s DB connection on exit.

_teardown() → None

Stop worker and listener threads; worker closes its own DB connection.