How to write documents to a Haystack document store

A Haystack document store is the boundary between prepared Document objects and later retrieval. Direct calls to write_documents() suit application code that already owns the store and does not need a DocumentWriter pipeline component.

Stable document IDs make the stored batch predictable across runs. DuplicatePolicy.FAIL protects those IDs by raising an error instead of silently replacing an existing document with different content.

An InMemoryDocumentStore keeps the storage run local and needs no database service. Calling filter_documents() after the write reconstructs each document's content and section metadata, allowing the stored state to be compared with the original batch rather than trusting only the returned write count.

1. Activate the project environment that contains haystack-ai.

   $ source .venv/bin/activate

   The project may use a virtual-environment path other than .venv.
   Related: How to install Haystack with pip

2. Create document_store_write_demo.py with the imports and input batch.

   document_store_write_demo.py

   from haystack import Document
   from haystack.document_stores.in_memory import InMemoryDocumentStore
   from haystack.document_stores.types import DuplicatePolicy
    
    
   documents = [
       Document(
           id="handbook-api",
           content="Use the client library to submit service requests.",
           meta={"section": "API"},
       ),
       Document(
           id="handbook-cli",
           content="Use the command line for scheduled maintenance.",
           meta={"section": "CLI"},
       ),
   ]

3. Add the expected content and metadata mapping below the input batch.

   document_store_write_demo.py

   expected_documents = {
       document.id: {
           "content": document.content,
           "section": document.meta["section"],
       }
       for document in documents
   }

4. Append the store initialization and guarded write below the expected mapping.

   document_store_write_demo.py

   document_store = InMemoryDocumentStore()
   written_count = document_store.write_documents(
       documents=documents,
       policy=DuplicatePolicy.FAIL,
   )

   The direct method returns the number written, while DuplicatePolicy.FAIL stops if either explicit ID already exists.
   Related: How to set duplicate document policy in Haystack

5. Read every stored document into a mapping below the write call.

   document_store_write_demo.py

   stored_documents = {
       document.id: {
           "content": document.content,
           "section": document.meta["section"],
       }
       for document in document_store.filter_documents()
   }

   Reading through the store protocol checks the retained records instead of reusing the original Document list.
   Related: How to filter documents in a Haystack document store

6. Append the count assertions and stored-document formatter below the read-back mapping.

   document_store_write_demo.py

   assert written_count == len(documents)
   assert stored_documents == expected_documents
    
   print(f"write_documents returned: {written_count}")
   print("verified stored documents:")
   for document_id in sorted(stored_documents):
       stored = stored_documents[document_id]
       print(f"- {document_id} [{stored['section']}]: {stored['content']}")

   The second assertion fails when an ID, body, section value, or stored record differs from the requested batch.
   Related: How to delete documents from a Haystack document store

7. Compare document_store_write_demo.py with the completed file.

   document_store_write_demo.py

   from haystack import Document
   from haystack.document_stores.in_memory import InMemoryDocumentStore
   from haystack.document_stores.types import DuplicatePolicy
    
    
   documents = [
       Document(
           id="handbook-api",
           content="Use the client library to submit service requests.",
           meta={"section": "API"},
       ),
       Document(
           id="handbook-cli",
           content="Use the command line for scheduled maintenance.",
           meta={"section": "CLI"},
       ),
   ]
    
   expected_documents = {
       document.id: {
           "content": document.content,
           "section": document.meta["section"],
       }
       for document in documents
   }
    
   document_store = InMemoryDocumentStore()
   written_count = document_store.write_documents(
       documents=documents,
       policy=DuplicatePolicy.FAIL,
   )
    
   stored_documents = {
       document.id: {
           "content": document.content,
           "section": document.meta["section"],
       }
       for document in document_store.filter_documents()
   }
    
   assert written_count == len(documents)
   assert stored_documents == expected_documents
    
   print(f"write_documents returned: {written_count}")
   print("verified stored documents:")
   for document_id in sorted(stored_documents):
       stored = stored_documents[document_id]
       print(f"- {document_id} [{stored['section']}]: {stored['content']}")

8. Run the completed program to verify the stored content and metadata.

   $ python document_store_write_demo.py
   write_documents returned: 2
   verified stored documents:
   - handbook-api [API]: Use the client library to submit service requests.
   - handbook-cli [CLI]: Use the command line for scheduled maintenance.