warren.jobs.status
Job status: an observer that records each document outcome and publishes
job-completed.
Classes
JobStatusWorker
classJobStatusWorker(worker_name, *, job_store, job_results_store, worker_type=None, store_retry_attempts=5, store_retry_base_delay=1.0)Observes messages on the observer exchange and records results.
Sits alongside processing workers. Classifies
each message (success, soft-failure, hard-failure) and records the
outcome via JobResultsStoreInterface. After each update, checks
whether the job is complete and returns a "job-completed"
message if so.
A message without data.doc_id is skipped: it does not count
toward completion. A job is complete when the successes at the
final_data_type of the job record, plus the hard failures, reach
num_documents. While num_documents is not set, the job does
not complete. The job-completed message has completed_count,
hard_failed_count and with_failures in data, and the
parameters of the job in job_parameters.
paramworker_namestrUnique instance identifier.
paramjob_storeJobStoreInterfaceStore for job definitions and completion status.
paramjob_results_storeJobResultsStoreInterfaceStore for per-document processing results.
paramworker_typestr | None= NoneOptional override for the worker type (default: "job_status_worker").
paramstore_retry_attemptsint= 5Number of attempts for each store call that raises TransientStoreError. After the last attempt, the observation is dropped.
paramstore_retry_base_delayfloat= 1.0Delay in seconds before the first retry of a store call. The delay grows with each attempt.
methodshould_process(message) -> boolProcess everything with a job_id, except own completion signals and job publication triggers.
"job-completed" is this worker's own output
(self-consumption guard). "job" is a publication trigger
for JobPublicationWorker, not a document-level processing
result. Publishing via the exchange is optional —
JobDocumentsPublisher can be used directly without RMQ —
so this worker must not depend on or react to it.
parammessagedictmethodprocessasync (message) -> dict | NoneClassify the message, record the result, and check completion.
Store interactions run under a bounded local retry on
TransientStoreError (run_with_transient_retry): this
worker is a pure observer, so a transient store blip that escaped
would drop the observation — losing a completion increment and
potentially stranding the job. Permanent errors and malformed
messages are not TransientStoreError, so they propagate
immediately (fail loud). A sustained outage exhausts the retries;
the observation is then dropped (acked) rather than re-fanned to
the bus.
parammessagedictJobStatusWorkerRunner
classJobStatusWorkerRunner(config, worker_name, *, exchange, job_store=None, job_results_store=None, consumer_manager_factory=None, infra=None)Runs a JobStatusWorker with its lifecycle hooks.
Creates infrastructure from RuntimeConfig and builds default
components in setup(). Inject custom components to override
defaults (e.g. for testing).
paramconfigRuntimeConfigruntime infrastructure configuration.
paramworker_namestrunique identifier for this worker instance.
paramexchangeRMQExchangeConfigThe observer exchange (observer_exchange(pipeline.exchange)). On a direct pipeline, the data exchange does not carry every message, so the worker sees no messages there.
paramjob_storeJobStoreInterface | None= Noneoptional override for the job definitions store. Default: MongoDBJobStore.
paramjob_results_storeJobResultsStoreInterface | None= Noneoptional override for the per-document results store. Default: MongoDBJobResultsStore.
paramconsumer_manager_factoryConsumerManagerFactory | None= Noneoptional override for the consumer manager factory. Default: the consumer manager of config.backend (from create_consumer_manager) on the job status queue, with a publisher of config.backend for the job-completed signal.
paraminfraRuntimeInfra | None= Noneoptional shared infrastructure. When given, the runner uses it and does not close it; the caller owns its lifetime. Needed when several runners share one process (backend: memory).
methodsetupasync () -> NoneCreate infrastructure, build defaults, wire the worker.
- Create infrastructure (MongoDB, Redis, pubsub connection)
- Build default stores and consumer factory for any not injected
- Create the JobStatusWorker
- Create and set up the consumer manager
Constants
JOB_STATUS_WORKER_TYPE
constantJOB_STATUS_WORKER_TYPE: str = 'job_status'