soev for developers
WarrenReferencewarren.storage

warren.storage.job_results

Job results: one status record for each document and stage of a job.

Classes

MemoryJobResultsStore

classMemoryJobResultsStore(*, name=None)

Per-document, per-stage status ledger held in a dict.

One record per (job_id, data_type, doc_id), the same identity the Mongo store enforces with its unique index. A record's failure state is carried by which fields are present: soft_failures (a list of reasons) or hard_failure (one error string). Recording an outcome removes the fields that belong to the other outcomes, exactly as the Mongo store does with $unset.

paramnamestr | None
= None
methodsetupasync () -> None

Nothing to create; present because the Protocol requires it.

methodrecord_successasync (job_id, data_type, doc_id, origin_type, origin_name) -> None
paramjob_idstr
paramdata_typestr
paramdoc_idstr
paramorigin_typestr
paramorigin_namestr
methodrecord_soft_failureasync (job_id, data_type, doc_id, origin_type, origin_name, consumer_type, consumer_name, retry_count, error_reason) -> None
paramjob_idstr
paramdata_typestr
paramdoc_idstr
paramorigin_typestr
paramorigin_namestr
paramconsumer_typestr
paramconsumer_namestr
paramretry_countint
paramerror_reasonstr
methodrecord_hard_failureasync (job_id, data_type, doc_id, origin_type, origin_name, consumer_type, consumer_name, error) -> None
paramjob_idstr
paramdata_typestr
paramdoc_idstr
paramorigin_typestr
paramorigin_namestr
paramconsumer_typestr
paramconsumer_namestr
paramerrorstr
methodcount_completed_docsasync (job_id, final_data_type) -> int
paramjob_idstr
paramfinal_data_typestr
methodcount_hard_failed_docsasync (job_id) -> int
paramjob_idstr
methodget_stage_countsasync (job_id) -> list[dict]
paramjob_idstr
methodget_doc_statusasync (job_id, doc_id) -> list[dict]
paramjob_idstr
paramdoc_idstr
methodget_failuresasync (job_id, data_type=None) -> list[dict]
paramjob_idstr
paramdata_typestr | None
= None
methodget_unique_errorsasync (job_id, data_type=None) -> list[dict]
paramjob_idstr
paramdata_typestr | None
= None

MongoDBJobResultsStore

classMongoDBJobResultsStore(client, *, database_name, collection_name='job_results', job_records_ttl_seconds=None, name=None, scoping_enabled=False)

Async MongoDB implementation of JobResultsStoreInterface.

Stores per-document processing results in a job_results collection. Each record is keyed by (job_id, data_type, doc_id) via a unique composite index, enabling atomic upsert operations.

Requires calling await setup() after construction to create indexes.

paramclientAsyncMongoClient
paramdatabase_namestr
paramcollection_namestr
= 'job_results'
paramjob_records_ttl_secondsint | None
= None

TTL on time: a record is deleted this many seconds after its last write. Must be positive. None: the index stays as it is (no TTL by default).

paramnamestr | None
= None
paramscoping_enabledbool
= False

Store the scope of the current handler on each record in a scope field, and index that field.

methodsetupasync () -> None

Create indexes for the job_results collection.

methodrecord_successasync (job_id, data_type, doc_id, origin_type, origin_name) -> None
paramjob_idstr
paramdata_typestr
paramdoc_idstr
paramorigin_typestr
paramorigin_namestr
methodrecord_soft_failureasync (job_id, data_type, doc_id, origin_type, origin_name, consumer_type, consumer_name, retry_count, error_reason) -> None
paramjob_idstr
paramdata_typestr
paramdoc_idstr
paramorigin_typestr
paramorigin_namestr
paramconsumer_typestr
paramconsumer_namestr
paramretry_countint
paramerror_reasonstr
methodrecord_hard_failureasync (job_id, data_type, doc_id, origin_type, origin_name, consumer_type, consumer_name, error) -> None
paramjob_idstr
paramdata_typestr
paramdoc_idstr
paramorigin_typestr
paramorigin_namestr
paramconsumer_typestr
paramconsumer_namestr
paramerrorstr
methodcount_completed_docsasync (job_id, final_data_type) -> int
paramjob_idstr
paramfinal_data_typestr
methodcount_hard_failed_docsasync (job_id) -> int
paramjob_idstr
methodget_stage_countsasync (job_id) -> list[dict]
paramjob_idstr
methodget_doc_statusasync (job_id, doc_id) -> list[dict]
paramjob_idstr
paramdoc_idstr
methodget_failuresasync (job_id, data_type=None) -> list[dict]
paramjob_idstr
paramdata_typestr | None
= None
methodget_unique_errorsasync (job_id, data_type=None) -> list[dict]
paramjob_idstr
paramdata_typestr | None
= None

Protocols

JobResultsStoreInterface

protocolJobResultsStoreInterface

Storage protocol for per-document processing results.

Transient-error contract: transient backend failures (connection blips, failover) are raised as warren.storage.TransientStoreError; permanent failures propagate as other exceptions. Callers that cannot tolerate a dropped operation — notably the JobStatusWorker observer, where a lost write strands a job — should retry on TransientStoreError.

methodsetupasync () -> None

Create indexes and prepare the store for use.

methodrecord_successasync (job_id, data_type, doc_id, origin_type, origin_name) -> None

Record a successful processing result.

Upserts the (job_id, data_type, doc_id) record. If the record previously held a failure (soft or hard), it is overwritten with success and failure-specific fields are cleared.

paramjob_idstr

Job ID. Part of the record key.

paramdata_typestr

The pipeline stage. Part of the record key.

paramdoc_idstr

Document ID. Part of the record key.

paramorigin_typestr

Who produced the message.

paramorigin_namestr

Name of the producer.

methodrecord_soft_failureasync (job_id, data_type, doc_id, origin_type, origin_name, consumer_type, consumer_name, retry_count, error_reason) -> None

Record a soft (retryable) failure.

Upserts the (job_id, data_type, doc_id) record. Appends error_reason to the soft_failures list so the full retry history is preserved.

paramjob_idstr

Job ID. Part of the record key.

paramdata_typestr

The pipeline stage. Part of the record key.

paramdoc_idstr

Document ID. Part of the record key.

paramorigin_typestr

Who produced the message that failed.

paramorigin_namestr

Name of the producer.

paramconsumer_typestr

Who failed to consume it.

paramconsumer_namestr

Name of the consumer.

paramretry_countint

Number of retries so far. Replaces the stored value.

paramerror_reasonstr

Reason for this failure.

methodrecord_hard_failureasync (job_id, data_type, doc_id, origin_type, origin_name, consumer_type, consumer_name, error) -> None

Record a hard (permanent) failure.

Upserts the (job_id, data_type, doc_id) record. This document will not be retried.

paramjob_idstr

Job ID. Part of the record key.

paramdata_typestr

The pipeline stage. Part of the record key.

paramdoc_idstr

Document ID. Part of the record key.

paramorigin_typestr

Who produced the message that failed.

paramorigin_namestr

Name of the producer.

paramconsumer_typestr

Who failed to consume it.

paramconsumer_namestr

Name of the consumer.

paramerrorstr

Error text, stored as the hard failure of the record.

methodcount_completed_docsasync (job_id, final_data_type) -> int

Count documents that reached the final stage successfully.

paramjob_idstr
paramfinal_data_typestr

Returns

int

Number of documents with success=True for final_data_type.

methodcount_hard_failed_docsasync (job_id) -> int

Count documents that hard-failed at any stage.

paramjob_idstr

Returns

int

Number of distinct doc_id values with a hard_failure record.

methodget_stage_countsasync (job_id) -> list[dict]

Get per-stage counts for progress display.

paramjob_idstr

Returns

list[dict]

List of dicts, each containing: {data_type, total, succeeded, soft_failed, hard_failed}.

methodget_doc_statusasync (job_id, doc_id) -> list[dict]

Get all stage records for a specific document.

paramjob_idstr
paramdoc_idstr

Returns

list[dict]

List of status records across all stages.

methodget_failuresasync (job_id, data_type=None) -> list[dict]

Get all failure records for a job, optionally filtered by stage.

paramjob_idstr
paramdata_typestr | None
= None

If provided, only return failures for this stage.

Returns

list[dict]

List of failure records (success=False).

methodget_unique_errorsasync (job_id, data_type=None) -> list[dict]

Group the failed records of a job by their current error.

The current error of a record is its hard failure, or else its last soft-failure reason. Earlier soft-failure reasons are not counted. count is the number of records (document and stage) in each group. first_seen and last_seen are the earliest and latest update time of those records.

paramjob_idstr
paramdata_typestr | None
= None

If provided, only return errors for this stage.

Returns

list[dict]

List of {error, count, data_type, first_seen, last_seen}.

warren.storage.documents

Documents: document locations, resolvers, the cached fetcher and source discovery.

warren.storage.jobs

Jobs: the job store, which holds what a job is and its completion status.

On this page

ClassesMemoryJobResultsStoreMongoDBJobResultsStoreProtocolsJobResultsStoreInterface