warren.storage.jobs
Jobs: the job store, which holds what a job is and its completion status.
Classes
MemoryJobStore
classMemoryJobStore(*, name=None, scoping_enabled=False)Job definitions and high-level status held in a dict.
Record shape, job id format and error types match MongoDBJobStore
so callers cannot tell the two apart. Single-process, gone at exit.
paramnamestr | None= Noneparamscoping_enabledbool= Falsemethodsetupasync () -> NoneNothing to create; present because the Protocol requires it.
methodcreate_jobasync (final_data_type, parameters=None, num_documents=None, metadata=None, scope=None) -> strparamfinal_data_typestrparamparametersdict | None= Noneparamnum_documentsint | None= Noneparammetadatadict | None= Noneparamscopestr | None= Nonemethodget_jobasync (job_id) -> dictparamjob_idstrmethodupdate_num_documentsasync (job_id, num_documents) -> Noneparamjob_idstrparamnum_documentsintmethodincrement_num_documentsasync (job_id, count=1) -> intparamjob_idstrparamcountint= 1methodupdate_completionasync (job_id, completed, with_failures) -> boolparamjob_idstrparamcompletedboolparamwith_failuresboolmethodget_statusasync (job_id) -> dictparamjob_idstrmethodupdate_vector_db_stateasync (job_id, fields) -> Noneparamjob_idstrparamfieldsdictMongoDBJobStore
classMongoDBJobStore(client, *, database_name, collection_name='jobs', job_records_ttl_seconds=None, job_records_max_age_seconds=None, name=None, scoping_enabled=False)Async MongoDB implementation of JobStoreInterface.
Stores job definitions and high-level status in a jobs
collection. Uses job_id as the unique key.
Requires calling await setup() after construction to create
indexes.
paramclientAsyncMongoClientparamdatabase_namestrparamcollection_namestr= 'jobs'paramjob_records_ttl_secondsint | None= NoneTTL on status.completed_at: a job record is deleted this many seconds after completion. Must be positive. None: the index stays as it is (no TTL by default).
paramjob_records_max_age_secondsint | None= NoneTTL on created_at: a job record is deleted this many seconds after creation. Must be positive. None: the index stays as it is (no TTL by default).
paramnamestr | None= Noneparamscoping_enabledbool= FalseStore the scope on each job in a scope field, and index that field.
methodsetupasync () -> NoneCreate job identity and retention indexes.
methodcreate_jobasync (final_data_type, parameters=None, num_documents=None, metadata=None, scope=None) -> strparamfinal_data_typestrparamparametersdict | None= Noneparamnum_documentsint | None= Noneparammetadatadict | None= Noneparamscopestr | None= Nonemethodget_jobasync (job_id) -> dictparamjob_idstrmethodupdate_num_documentsasync (job_id, num_documents) -> Noneparamjob_idstrparamnum_documentsintmethodincrement_num_documentsasync (job_id, count=1) -> intparamjob_idstrparamcountint= 1methodupdate_completionasync (job_id, completed, with_failures) -> boolparamjob_idstrparamcompletedboolparamwith_failuresboolmethodget_statusasync (job_id) -> dictparamjob_idstrmethodupdate_vector_db_stateasync (job_id, fields) -> Noneparamjob_idstrparamfieldsdictProtocols
JobStoreInterface
protocolJobStoreInterfaceStorage protocol for job definitions and job-level status.
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 completion update
strands a job — should retry on TransientStoreError.
methodsetupasync () -> NoneCreate indexes and prepare the store for use.
methodcreate_jobasync (final_data_type, parameters=None, num_documents=None, metadata=None, scope=None) -> strCreate a new job entry with a store-generated ID.
paramfinal_data_typestrThe data_type that marks a document as fully processed (e.g., "final_stage").
paramparametersdict | None= NoneProcessing parameters for workers.
paramnum_documentsint | None= NoneTotal document count, if known upfront. Can be set later via update_num_documents.
parammetadatadict | None= NoneOptional application-defined metadata.
paramscopestr | None= NoneScope to store on the job; None uses the current handler scope. Ignored when scoping is off or the scope is not valid.
Returns
strThe generated job ID.
Raises:
JobCreationFailedError: If the store cannot create the job.
methodget_jobasync (job_id) -> dictGet the full job document.
paramjob_idstrReturns
dictJob document as dict with all fields.
Raises:
JobNotFoundError: If job_id does not exist.
methodupdate_num_documentsasync (job_id, num_documents) -> NoneSet or update the number of documents for a job.
A value of 0 means "no documents were successfully submitted"
— different from None, which means "not yet known."
paramjob_idstrparamnum_documentsintRaises:
JobNotFoundError: If job_id does not exist.
methodincrement_num_documentsasync (job_id, count=1) -> intAtomically increment num_documents and return the new value.
Used when documents are published in batches or streamed, and the total is not known upfront.
paramjob_idstrparamcountint= 1Returns
intNew num_documents value after increment.
Raises:
JobNotFoundError: If job_id does not exist.
methodupdate_completionasync (job_id, completed, with_failures) -> boolUpdate an unfinished job; return whether this call applied the update.
A completion timestamp prevents later updates, including reopening.
paramjob_idstrparamcompletedboolparamwith_failuresboolRaises:
JobNotFoundError: If job_id does not exist.
methodget_statusasync (job_id) -> dictGet the job's current status.
paramjob_idstrReturns
dictDict with {completed, with_failures, completed_at}.
Raises:
JobNotFoundError: If job_id does not exist.
methodupdate_vector_db_stateasync (job_id, fields) -> NonePartially update the job's vector_db sub-document.
A worker that tracks per-job external state uses it to record state transitions. Idempotent — repeated calls with the same fields are safe.
paramjob_idstrJob to update.
paramfieldsdictKeys inside the vector_db sub-document, without a prefix, and their values. E.g. {"status": "ready"}. An empty mapping does nothing.
Raises:
JobNotFoundError: If job_id does not exist.
Errors
JobCreationFailedError
errorJobCreationFailedErrorThe store failed to create the job.
JobNotFoundError
errorJobNotFoundError