warren.storage.document_store
Document store: the document store interface and its MongoDB and in-memory implementations.
Classes
MemoryDocumentStore
classMemoryDocumentStore(*, collection_name, doc_id_field='doc_id', unique_indexes=None, name=None)Document store held in a dict. Single-process, gone at exit.
Takes the same doc_id_field / unique_indexes keywords as
MongoDBDocumentStore so either can sit under DefaultResultsStore
and RetryWorker. The default doc_id_field is "doc_id", not
"_id". Pass it explicitly when the two stores must be
interchangeable.
Documents are deep-copied on the way in and on the way out. A database gives callers that isolation for free; a dict does not, and a worker mutating a document it just read must not change stored state.
The doc id field and each unique index are backed by a hash index, so
uniqueness is checked in constant time instead of by scanning the
collection. Values of indexed fields must be hashable. As on MongoDB,
an update that would duplicate a unique key is rejected, here with
DocumentAlreadyExistsError.
query supports flat equality only, which is everything the
framework itself issues. A query operator raises instead of silently
matching nothing.
paramcollection_namestrparamdoc_id_fieldstr= 'doc_id'paramunique_indexeslist[IndexSpec] | None= Noneparamnamestr | None= Nonemethodinsertasync (doc, overwrite_existing=False) -> strparamdocdictparamoverwrite_existingbool= Falsemethodupsert_manyasync (docs) -> list[str | None]paramdocslist[dict]methodupdateasync (doc_id, updates) -> Noneparamdoc_idstrparamupdatesdictmethodexistsasync (doc_id) -> boolparamdoc_idstrmethodget_documentasync (doc_id) -> dictparamdoc_idstrmethodqueryasync (params, *, sort_by=None) -> AsyncGenerator[dict, None]paramparamsdictparamsort_bystr | None= Nonemethoddeleteasync (doc_id, *, expected=None) -> boolparamdoc_idstrparamexpecteddict | None= Nonemethodget_document_type() -> strmethodget_doc_id_field() -> strmethodhas_unique_indexasync (index_spec) -> boolparamindex_specIndexSpecMongoDBDocumentStore
classMongoDBDocumentStore(client, *, database_name, collection_name, doc_id_field='_id', fields_to_index=None, unique_indexes=None, name=None, scoped_database=None)Async MongoDB implementation of DocumentStoreInterface.
Uses pymongo's AsyncMongoClient for non-blocking I/O. Provides document storage using MongoDB as the backend. Supports custom document ID fields and automatic index creation.
Requires calling await setup() after construction to create indexes.
create_default_results_store and create_binary_results_store call
it for the store that they create.
Initialize the MongoDB document store.
paramclientAsyncMongoClientAsync MongoDB client instance.
paramdatabase_namestrName of the database to use.
paramcollection_namestrName of the collection to use.
paramdoc_id_fieldstr= '_id'Field name to use as document ID. Defaults to "_id".
paramfields_to_indexlist[IndexSpec] | None= NoneList of index specifications. Each can be a single field name (str) or a tuple of field names for composite indexes.
paramunique_indexeslist[IndexSpec] | None= NoneList of index specifications that should be unique. Each can be a single field name (str) or a tuple of field names.
paramnamestr | None= NoneOptional name for logging purposes.
paramscoped_databaseScopedDatabase | None= NoneResolve the database from the scope of the current handler (database_name is then not used). The indexes of each scope database are created when the scope is first used, and setup() does nothing.
methodsetupasync () -> NoneCreate indexes on configured fields.
Must be called after construction before using the store.
create_default_results_store and create_binary_results_store
call it for the store that they create.
methodinsertasync (doc, overwrite_existing=False) -> strInsert a document into the document store.
paramdocdictDocument to insert.
paramoverwrite_existingbool= FalseIf True, replace existing document with same unique key. If unique_indexes are configured, uses the first unique index as the upsert key. Otherwise, uses the doc_id_field.
Returns
strDocument ID of the inserted document.
Raises:
DocumentAlreadyExistsError: If overwrite_existing is False and the document has the same ID or the same values for a unique index as a stored document.ValueError: If the ID is empty, or is not a valid ObjectId when doc_id_field is "_id".
methodupsert_manyasync (docs) -> list[str | None]Upsert rows in one unordered bulk write, preserving MongoDB IDs.
paramdocslist[dict]methodupdateasync (doc_id, updates) -> NoneUpdate a document with the given updates.
paramdoc_idstrID of the document to update.
paramupdatesdictDictionary of field updates to apply.
Raises:
ValueError: If doc_id is None, empty, or invalid format.DocumentNotFoundError: If document with doc_id does not exist.
methodexistsasync (doc_id) -> boolCheck if a document exists in the store.
paramdoc_idstrID of the document to check.
Returns
boolTrue if document exists, False otherwise.
Raises:
ValueError: If doc_id is None, empty, or invalid format.
methoddeleteasync (doc_id, *, expected=None) -> boolDelete a document by its ID.
paramdoc_idstrID of the document to delete.
paramexpecteddict | None= NoneAdditional field values required for deletion.
Returns
boolTrue if document existed and was deleted, False if document was not found.
Raises:
ValueError: If doc_id is None, empty, or invalid format.
methodget_documentasync (doc_id) -> dictRetrieve a document by its ID.
paramdoc_idstrID of the document to retrieve.
Returns
dictThe document as a dictionary.
Raises:
ValueError: If doc_id is None, empty, or invalid format.DocumentNotFoundError: If document with doc_id does not exist.
methodqueryasync (params, *, sort_by=None) -> AsyncGenerator[dict, None]Query documents matching the given parameters.
paramparamsdictMongoDB query parameters.
paramsort_bystr | None= NoneOptional field to sort by in ascending order.
Returns
AsyncGenerator[dict, None]Async generator yielding matching documents.
methodget_document_type() -> strReturns the document type identifier for this store.
Returns
strThe collection name.
methodget_doc_id_field() -> strReturns the field name used for document IDs.
Returns
strThe configured doc_id_field.
methodhas_unique_indexasync (index_spec) -> boolCheck if a unique index exists on the given field(s).
paramindex_specIndexSpecSingle field name or tuple of field names for composite index.
Returns
boolTrue if a unique index exists on exactly these fields.
Protocols
DocumentStoreInterface
protocolDocumentStoreInterfaceAsync document store interface.
All I/O methods are async — implementations must use non-blocking drivers (e.g., pymongo.AsyncMongoClient) to avoid blocking the event loop.
Accessor methods (get_document_type, get_doc_id_field) are sync because they return stored config values with no I/O.
methodinsertasync (doc, overwrite_existing=False) -> strInsert a document into the document store and return its ID.
paramdocdictDocument to insert.
paramoverwrite_existingbool= FalseAllow overwrite of existing document if True.
Returns
strDocument ID.
Raises:
DocumentAlreadyExistsError: If overwrite_existing is False and the document has the same ID or the same values for a unique index as a stored document.ValueError: If the ID is empty, or is not a valid ObjectId when doc_id_field is "_id".
methodupsert_manyasync (docs) -> list[str | None]Replace rows by unique key, returning IDs where the backend provides them.
paramdocslist[dict]methodupdateasync (doc_id, updates) -> NoneUpdate document with document id with given updates.
paramdoc_idstrDocument ID.
paramupdatesdictFields to set. Fields that are not named stay as they are. Use top-level keys only: a dotted key is a nested path on MongoDB and a literal key in memory.
Raises:
DocumentNotFoundError: if document id does not exist.
methodexistsasync (doc_id) -> boolCheck if document id exists.
paramdoc_idstrDocument ID.
Returns
boolTrue if document id exists.
methodget_documentasync (doc_id) -> dictGet document with document id.
paramdoc_idstrDocument ID.
Returns
dictDocument dict.
Raises:
DocumentNotFoundError: if document id does not exist.
methodquery(params, *, sort_by=None) -> AsyncGenerator[dict, None]Query document store.
Returns an async generator — callers use async for doc in store.query(params).
paramparamsdictQuery parameters.
paramsort_bystr | None= NoneOptional field to sort by in ascending order.
Returns
AsyncGenerator[dict, None]Async generator of documents found.
methoddeleteasync (doc_id, *, expected=None) -> boolDelete a document by its ID.
paramdoc_idstrDocument ID.
paramexpecteddict | None= NoneAdditional field values required for deletion.
Returns
boolTrue if document existed and was deleted, False if it was not found or expected values did not match.
methodget_document_type() -> strReturns the document type identifier for this store.
Returns
strDocument type identifier (e.g., collection name).
methodget_doc_id_field() -> strReturns the field name used for document IDs.
Returns
strField name containing document IDs.
methodhas_unique_indexasync (index_spec) -> boolCheck if a unique index exists on the given field(s).
paramindex_specIndexSpecSingle field name or tuple of field names for composite index.
Returns
boolTrue if a unique index exists on exactly these fields.
Type aliases
DocId
type aliasDocId = Union[ObjectId, str]IndexSpec
type aliasIndexSpec = Union[str, tuple[str, ...]]Errors
DocumentAlreadyExistsError
errorDocumentAlreadyExistsErrorDocumentNotFoundError
errorDocumentNotFoundErrorNo document with the given ID exists in the document store.
Raised by document stores. Resolvers and the fetcher raise a different
class, warren.storage.documents.DocumentNotFoundError. A handler that
catches one class does not catch the other.