ursa.status

Read-only ursa.status.* namespace — catalog observability.

The per-package *.status.* observability convention: a read-only namespace returning Pydantic models with to_dataframe() / to_markdown(). This is the single per-recording ingestion-status contract consumed by Cepheus’s /data page and the constellation-status-api HTTP wrapper — the recording-level aggregation lives here so consumers don’t each re-derive the precedence rules.

Built over the catalog’s ingestion-lifecycle columns (ingestion_status + ingestion_error / ingestion_attempts / last_attempt_at + the durable metadata["dead_letter"] flag). Read-only: no writes, no side effects.

Module Contents

Classes

ActiveJobs	In-flight ingestion work — modalities currently leased (ingesting).
ModalityIngestion	One modality’s ingestion state (a flattened view of its ModalityRow).
RecentIngestions	Result of :func:recent_ingestions. rows is the (limited) recent slice; the n_* counts are over the full post-since set (pre-limit).
RecordingIngestion	Recording-level ingestion status, aggregated across all its modalities.

Functions

active_jobs	Modalities currently being ingested (ingestion_status="ingesting") — the in-flight leases held by the Virgo ingestion daemon.
aggregate_recording_ingestion	Aggregate a mixed-status QueryResult (from query(status="all")) into a single recording-level ingestion status.
recent_ingestions	Per-recording ingestion status, most-recently-attempted first.

API

class ursa.status.ActiveJobs(/, **data: typing.Any)

Bases: pydantic.BaseModel

In-flight ingestion work — modalities currently leased (ingesting).

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

‘ConfigDict(…)’

ingesting: tuple[ursa.status.ModalityIngestion, ...]

None

n_ingesting: int

None

class ursa.status.ModalityIngestion(/, **data: typing.Any)

Bases: pydantic.BaseModel

One modality’s ingestion state (a flattened view of its ModalityRow).

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

‘ConfigDict(…)’

modality: str

None

worker_id: str

None

status: ursa.catalog.IngestionStatus

None

error: str | None

None

attempts: int

None

last_attempt_at: int | None

None

class ursa.status.RecentIngestions(/, **data: typing.Any)

Bases: pydantic.BaseModel

Result of :func:recent_ingestions. rows is the (limited) recent slice; the n_* counts are over the full post-since set (pre-limit).

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

‘ConfigDict(…)’

rows: tuple[ursa.status.RecordingIngestion, ...]

None

n_processed: int

None

n_failed: int

None

n_ingesting: int

None

n_raw: int

None

to_dataframe() → pandas.DataFrame

to_markdown() → str

class ursa.status.RecordingIngestion(/, **data: typing.Any)

Bases: pydantic.BaseModel

Recording-level ingestion status, aggregated across all its modalities.

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

‘ConfigDict(…)’

recording_hash: str

None

status: ursa.catalog.IngestionStatus

None

dead_letter: bool

None

error: str | None

None

started_at: int

None

last_attempt_at: int | None

None

modalities: tuple[ursa.status.ModalityIngestion, ...]

None

ursa.status.active_jobs(*, profile: str = 'r2', data: ursa.data_interface.DataInterface | None = None) → ursa.status.ActiveJobs

Modalities currently being ingested (ingestion_status="ingesting") — the in-flight leases held by the Virgo ingestion daemon.

ursa.status.aggregate_recording_ingestion(qr: ursa._query_types.QueryResult) → ursa.status.RecordingIngestion

Aggregate a mixed-status QueryResult (from query(status="all")) into a single recording-level ingestion status.

Precedence: failed > ingesting > raw > processed — the recording reads failed if any worker failed. dead_letter is true if any failed modality carries the durable metadata["dead_letter"] flag; error is the first failed modality’s message. A recording with no modality rows aggregates to raw (registered but never ingested) — never processed.

expanded rows (combined modalities split into per-stream children) rank alongside processed and fold to processed in the aggregate, so the recording-level status is never expanded (Cepheus only knows the four pre-existing statuses). They are also omitted from the per-modality modalities tuple — the data lives in their children (themselves processed rows in the same result).

ursa.status.recent_ingestions(*, profile: str = 'r2', since: int | None = None, limit: int | None = 50, data: ursa.data_interface.DataInterface | None = None) → ursa.status.RecentIngestions

Per-recording ingestion status, most-recently-attempted first.

Queries the catalog once with status="all" and aggregates each recording via :func:aggregate_recording_ingestion. since (epoch-ns) keeps only recordings whose last_attempt_at is set and >= since — this drops never-attempted raw rows and any pre-migration processed rows whose last_attempt_at was backfilled to NULL by migrate_modalities_ingestion_state. limit truncates rows (pass None for the full list, as Cepheus’s recordings page does). The n_* counts cover the full post-since set (the same windowed scope as rows, not just the limit-truncated slice).