warren.storage.results
Results: worker outputs keyed by document, job and part, with an optional cache.
Classes
BinaryResultsStore
classBinaryResultsStore(document_store, cache=None, result_type=None, name=None, *, cache_ttl_seconds=3600)Results store for byte payloads.
Focused API — intentionally does not implement
ResultsStoreInterface because that Protocol's result: Dict
contract doesn't fit byte payloads (and broadening it to Any
would ripple through every reader).
paramdocument_storeDocumentStoreInterfaceMongoDB-backed store for persistence. Must have a unique composite index on (doc_id, job_id, part_idx) (validated in setup()).
paramcacheCacheInterface[bytes] | None= NoneOptional CacheInterface[bytes] for fast reads. When absent, every read hits MongoDB.
paramresult_typestr | None= NoneNot used. Give each results store its own cache base_key. Two stores that share one cache instance overwrite each other's entries.
paramnamestr | None= NoneOptional logger name.
paramcache_ttl_secondsint= 3600Expiry for cached payloads.
methodsetupasync () -> NoneValidate the document store's unique composite index.
methodstore_bytesasync (payload, *, doc_id, part_idx=0, job_id=None, metadata=None, do_cache=True, overwrite_existing=True) -> strPersist payload and populate the cache.
The cache key is doc:<doc_id>:<job_id>, or doc:<doc_id>
when job_id is None. This is the key that
CachedDocumentFetcher reads.
parampayloadbytesThe bytes to store.
paramdoc_idstrDocument ID.
parampart_idxint= 0Part index. 0 for a single-part payload.
paramjob_idstr | None= NoneJob ID, or None if you do not group by job.
parammetadatadict[str, Any] | None= NoneOptional metadata, stored in the field metadata (DefaultResultsStore uses result_metadata).
paramdo_cachebool= TrueWhether to write the payload to the cache.
paramoverwrite_existingbool= TrueWhether to replace a stored record with the same (doc_id, job_id, part_idx).
Returns
strResult ID from the document store.
Raises:
DocumentAlreadyExistsError: Ifoverwrite_existingis False and a record for these keys exists.
methodget_bytesasync (doc_id, *, part_idx=0, job_id=None) -> bytesRead bytes by business keys. Cache-first, then MongoDB.
paramdoc_idstrparampart_idxint= 0paramjob_idstr | None= NoneRaises:
ResultNotFound: If no record exists for the given keys.
methodquery_metadataasync (*, doc_id, job_id=None) -> AsyncGenerator[dict[str, Any], None]Yield the stored records of a document and job, without their payload.
Use it to list records without the transfer of the payload bytes.
paramdoc_idstrparamjob_idstr | None= NoneDefaultResultsStore
classDefaultResultsStore(document_store, cache=None, result_type=None, name=None, *, cache_ttl_seconds=3600)Default async implementation of ResultsStoreInterface.
Stores document processing results using an injected DocumentStoreInterface for persistence and an optional CacheInterface for caching.
The document store must have a unique composite index on (doc_id, job_id, part_idx)
to ensure uniqueness and enable efficient queries. The setup() method validates
this requirement and raises ValueError if the index is missing.
Call await setup() after construction. create_default_results_store
calls it for you.
Initialize the results store.
paramdocument_storeDocumentStoreInterfaceStorage backend for persistence. Must have a unique composite index on (doc_id, job_id, part_idx).
paramcacheCacheInterface[dict] | None= NoneOptional cache for read-through/write-through caching.
paramresult_typestr | None= NoneNot used. Give each results store its own cache base_key. Two stores that share one cache instance overwrite each other's entries.
paramnamestr | None= NoneOptional name for logging purposes.
paramcache_ttl_secondsint= 3600Expiry for cached results.
methodsetupasync () -> NoneValidate that the document store has the required unique index.
Call it after construction, before you use the store.
create_default_results_store calls it for you.
Raises:
ValueError: If document store lacks required unique index.
methodstoreasync (result, doc_id, part_idx=None, job_id=None, result_metadata=None, do_cache=True, overwrite_existing=True) -> strStore a processing result.
paramresultdictThe processing result to store.
paramdoc_idstrDocument ID.
parampart_idxint | None= NonePart index (None or 0 for single-part results).
paramjob_idstr | None= NoneJob ID (None if not using job grouping).
paramresult_metadatadict | None= NoneOptional metadata about how the result was produced.
paramdo_cachebool= TrueWhether to cache the result.
paramoverwrite_existingbool= TrueWhether to overwrite existing result.
Returns
strDocument ID from the document store.
Raises:
DocumentAlreadyExistsError: If overwrite_existing is False and a result for these keys exists.
methodstore_manyasync (items) -> NoneUpsert results in bulk. For repeated keys, the last item wins.
paramitemsSequence[ResultItem]methodget_resultasync (doc_id, part_idx=None, job_id=None) -> ResultDocRetrieve a processing result by business keys.
paramdoc_idstrDocument ID.
parampart_idxint | None= NonePart index (None or 0 for single-part results).
paramjob_idstr | None= NoneJob ID (None if not using job grouping).
Returns
ResultDocThe result document.
Raises:
ResultNotFound: If no result exists for the given keys.
methodstream_doc_processing_resultsasync (doc_id, job_id=None, try_cache=True) -> AsyncGenerator[ResultDoc, None]Stream persisted results in ascending part order, bypassing the cache.
paramdoc_idstrDocument ID.
paramjob_idstr | None= NoneJob ID (None if not using job grouping).
paramtry_cachebool= TrueDeprecated and ignored; retained for compatibility.
Returns
AsyncGenerator[ResultDoc, None]Async generator yielding result documents.
Raises:
DocumentProcessingResultsNotFound: If no results exist.
ResultDoc
modelResultDocDocument representing a stored processing result.
attributedoc_idstrattributepart_idxint= 0attributejob_idstr | None= Noneattributeresultdictattributeresult_metadatadict | None= Noneattributecreated_atdatetime | None= Noneattributeresult_idstr | None= NoneResultItem
dataclassResultItem(result, doc_id, part_idx=None, job_id=None, result_metadata=None)A processing result and its storage keys.
paramresultdictparamdoc_idstrparampart_idxint | None= Noneparamjob_idstr | None= Noneparamresult_metadatadict | None= NoneProtocols
ResultsStoreInterface
protocolResultsStoreInterfaceAsync document-level results store for a specific type of document processing.
All methods are async — implementations must use non-blocking I/O to avoid blocking the event loop.
Assumes storage and caching capabilities that are aware of the specific result type. The implementation is dependent on the result/worker type.
methodstoreasync (result, doc_id, part_idx=None, job_id=None, result_metadata=None, do_cache=True, overwrite_existing=True) -> strStore (partial) result for processing a document.
paramresultdictDocument processing (partial) result.
paramdoc_idstrOriginal document ID.
parampart_idxint | None= NonePart index of the result (None or 0 for single-part results).
paramjob_idstr | None= NoneJob ID of the processing job (None if not using job grouping).
paramresult_metadatadict | None= NoneMetadata about how the result was produced.
paramdo_cachebool= TrueWhether to cache the result.
paramoverwrite_existingbool= TrueWhether to overwrite existing result.
Returns
strID of the stored result.
methodstore_manyasync (items) -> NoneUpsert results in bulk. For repeated keys, the last item wins.
paramitemsSequence[ResultItem]methodget_resultasync (doc_id, part_idx=None, job_id=None) -> ResultDocRetrieves the document processing result for given business keys.
paramdoc_idstrThe document ID.
parampart_idxint | None= NoneThe part index (None or 0 for single-part results).
paramjob_idstr | None= NoneThe job ID (None if not using job grouping).
Returns
ResultDocThe result document.
Raises:
ResultNotFound: If no result exists for the given keys.
methodstream_doc_processing_results(doc_id, job_id=None, try_cache=True) -> AsyncGenerator[ResultDoc, None]Stream persisted results in ascending part order, bypassing the cache.
paramdoc_idstrThe document ID.
paramjob_idstr | None= NoneThe job ID (None if not using job grouping).
paramtry_cachebool= TrueDeprecated and ignored; retained for compatibility.
Returns
AsyncGenerator[ResultDoc, None]Async generator yielding result documents.
Raises:
DocumentProcessingResultsNotFound: If no results exist for the document and job.
Functions
create_binary_results_store
functioncreate_binary_results_storeasync (collection_name, mongo_client, redis_client=None, *, database_name='document_processing', doc_id_field='result_id', cache_base_key='documents', cache_ttl_seconds=3600, scoped_database=None) -> BinaryResultsStoreCreate a BinaryResultsStore wired with MongoDB + RedisBinaryCache.
paramcollection_namestrMongoDB collection name.
parammongo_clientAsyncMongoClientAsync MongoDB client.
paramredis_clientRedis | None= NoneAsync Redis client. None disables caching.
paramdatabase_namestr= 'document_processing'MongoDB database name.
paramdoc_id_fieldstr= 'result_id'Field name for the per-result primary key.
paramcache_base_keystr= 'documents'Redis namespace. The default "documents" is the namespace that CachedDocumentFetcher reads. So a downstream worker that reads through GetDocumentFunc gets these bytes from the cache.
paramcache_ttl_secondsint= 3600Expiry for cached payloads in seconds (default 3600).
paramscoped_databaseScopedDatabase | None= NoneResolve the database from the scope of the current handler (database_name is then not used), and scope the cache keys. The indexes of a scope database are created when the scope is first used.
create_default_results_store
functioncreate_default_results_storeasync (collection_name, mongo_client, redis_client=None, *, database_name='document_processing', doc_id_field='result_id', cache_base_key_prefix='results', cache_ttl_seconds=3600, scoped_database=None) -> DefaultResultsStoreCreate a DefaultResultsStore with MongoDB persistence and optional Redis cache.
Returns a fully initialized store (setup() already called on both the document store and the results store).
paramcollection_namestrMongoDB collection name.
parammongo_clientAsyncMongoClientInjected async MongoClient instance.
paramredis_clientRedis | None= NoneInjected async Redis instance. If None, no caching.
paramdatabase_namestr= 'document_processing'MongoDB database name.
paramdoc_id_fieldstr= 'result_id'Field name for document IDs.
paramcache_base_key_prefixstr= 'results'Prefix for Redis cache keys. The cache base_key is <cache_base_key_prefix>:<collection_name>.
paramcache_ttl_secondsint= 3600Default TTL for cached results.
paramscoped_databaseScopedDatabase | None= NoneResolve the database from the scope of the current handler (database_name is then not used), and scope the cache keys. The indexes of a scope database are created when the scope is first used.
Errors
DocumentProcessingResultsNotFound
errorDocumentProcessingResultsNotFoundResultNotFound
errorResultNotFound