import json
from abc import ABC, abstractmethod
from typing import List, Optional
from swarmauri.core.documents.IDocument import IDocument
from swarmauri.core.document_stores.IDocumentStore import IDocumentStore
[docs]
class DocumentStoreBase(IDocumentStore, ABC):
"""
Abstract base class for document stores, implementing the IDocumentStore interface.
This class provides a standard API for adding, updating, getting, and deleting documents in a store.
The specifics of storing (e.g., in a database, in-memory, or file system) are to be implemented by concrete subclasses.
"""
[docs]
@abstractmethod
def add_document(self, document: IDocument) -> None:
"""
Add a single document to the document store.
Parameters:
- document (IDocument): The document to be added to the store.
"""
pass
[docs]
@abstractmethod
def add_documents(self, documents: List[IDocument]) -> None:
"""
Add multiple documents to the document store in a batch operation.
Parameters:
- documents (List[IDocument]): A list of documents to be added to the store.
"""
pass
[docs]
@abstractmethod
def get_document(self, doc_id: str) -> Optional[IDocument]:
"""
Retrieve a single document by its identifier.
Parameters:
- doc_id (str): The unique identifier of the document to retrieve.
Returns:
- Optional[IDocument]: The requested document if found; otherwise, None.
"""
pass
[docs]
@abstractmethod
def get_all_documents(self) -> List[IDocument]:
"""
Retrieve all documents stored in the document store.
Returns:
- List[IDocument]: A list of all documents in the store.
"""
pass
[docs]
@abstractmethod
def update_document(self, doc_id: str, updated_document: IDocument) -> None:
"""
Update a document in the document store.
Parameters:
- doc_id (str): The unique identifier of the document to update.
- updated_document (IDocument): The updated document instance.
"""
pass
[docs]
@abstractmethod
def delete_document(self, doc_id: str) -> None:
"""
Delete a document from the document store by its identifier.
Parameters:
- doc_id (str): The unique identifier of the document to delete.
"""
pass
[docs]
def document_count(self):
return len(self.documents)
[docs]
def dump(self, file_path):
with open(file_path, 'w') as f:
json.dumps([each.__dict__ for each in self.documents], f, indent=4)
[docs]
def load(self, file_path):
with open(file_path, 'r') as f:
self.documents = json.loads(f)