How to write documents with DocumentWriter in Haystack

Processed content becomes searchable only after its Document objects reach a document store. DocumentWriter handles that final indexing handoff in Haystack by accepting a document list and sending it through the store's supported write interface.

An InMemoryDocumentStore keeps this first write inside one Python process, so no database service or credentials are needed. Its contents disappear when the process exits; applications that need durable indexes should replace it with a persistent store supported by Haystack.

Stable IDs make the three records identifiable during the final read-back. DuplicatePolicy.FAIL also stops the run if an ID already exists, and the assertions check both the writer result and the records retrieved from the store.

1. Create document_writer_demo.py with the Haystack imports and an in-memory document store.

   document_writer_demo.py

   from haystack import Document
   from haystack.components.writers import DocumentWriter
   from haystack.document_stores.in_memory import InMemoryDocumentStore
   from haystack.document_stores.types import DuplicatePolicy
    
    
   document_store = InMemoryDocumentStore()

   Related: How to install Haystack with pip

2. Add three stable-ID records below the document_store assignment.

   document_writer_demo.py

   incoming_documents = [
       Document(
           id="billing-policy",
           content="Refunds are available for paid plans within 14 days.",
           meta={"source": "billing.md"},
       ),
       Document(
           id="support-hours",
           content="Support is available every weekday from 09:00 to 17:00.",
           meta={"source": "support.md"},
       ),
       Document(
           id="shipping-region",
           content="Standard shipping is available in the United States and Canada.",
           meta={"source": "shipping.md"},
       ),
   ]

3. Configure DocumentWriter to reject an ID that already exists in the store.

   document_writer_demo.py

   writer = DocumentWriter(
       document_store=document_store,
       policy=DuplicatePolicy.FAIL,
   )

   DuplicatePolicy.FAIL raises an error instead of silently skipping or replacing a stored ID.
   Related: How to set duplicate document policy in Haystack

4. Write the incoming records through the configured component.

   document_writer_demo.py

   write_result = writer.run(documents=incoming_documents)

5. Add a read-back block that validates the writer count and stored records.

   document_writer_demo.py

   stored_documents = sorted(
       document_store.filter_documents(),
       key=lambda document: document.id,
   )
    
   print(f"documents written: {write_result['documents_written']}")
   print(f"documents stored: {document_store.count_documents()}")
   for document in stored_documents:
       print(f"stored: {document.id} | source={document.meta['source']}")
    
   assert write_result["documents_written"] == len(incoming_documents)
   assert document_store.count_documents() == len(incoming_documents)
   assert [(document.id, document.meta["source"]) for document in stored_documents] == [
       ("billing-policy", "billing.md"),
       ("shipping-region", "shipping.md"),
       ("support-hours", "support.md"),
   ]

6. Compare document_writer_demo.py with the consolidated program.

   document_writer_demo.py

   from haystack import Document
   from haystack.components.writers import DocumentWriter
   from haystack.document_stores.in_memory import InMemoryDocumentStore
   from haystack.document_stores.types import DuplicatePolicy
    
    
   document_store = InMemoryDocumentStore()
    
   incoming_documents = [
       Document(
           id="billing-policy",
           content="Refunds are available for paid plans within 14 days.",
           meta={"source": "billing.md"},
       ),
       Document(
           id="support-hours",
           content="Support is available every weekday from 09:00 to 17:00.",
           meta={"source": "support.md"},
       ),
       Document(
           id="shipping-region",
           content="Standard shipping is available in the United States and Canada.",
           meta={"source": "shipping.md"},
       ),
   ]
    
   writer = DocumentWriter(
       document_store=document_store,
       policy=DuplicatePolicy.FAIL,
   )
   write_result = writer.run(documents=incoming_documents)
    
   stored_documents = sorted(
       document_store.filter_documents(),
       key=lambda document: document.id,
   )
    
   print(f"documents written: {write_result['documents_written']}")
   print(f"documents stored: {document_store.count_documents()}")
   for document in stored_documents:
       print(f"stored: {document.id} | source={document.meta['source']}")
    
   assert write_result["documents_written"] == len(incoming_documents)
   assert document_store.count_documents() == len(incoming_documents)
   assert [(document.id, document.meta["source"]) for document in stored_documents] == [
       ("billing-policy", "billing.md"),
       ("shipping-region", "shipping.md"),
       ("support-hours", "support.md"),
   ]

7. Run the completed program to verify the writer-to-store handoff.

   $ python document_writer_demo.py
   documents written: 3
   documents stored: 3
   stored: billing-policy | source=billing.md
   stored: shipping-region | source=shipping.md
   stored: support-hours | source=support.md