Spaces:
Runtime error
Runtime error
File size: 1,822 Bytes
35b22df |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
"""Mongo client."""
from typing import Dict, List, Optional
from gpt_index.readers.base import BaseReader
from gpt_index.readers.schema.base import Document
class SimpleMongoReader(BaseReader):
"""Simple mongo reader.
Concatenates each Mongo doc into Document used by LlamaIndex.
Args:
host (str): Mongo host.
port (int): Mongo port.
max_docs (int): Maximum number of documents to load.
"""
def __init__(self, host: str, port: int, max_docs: int = 1000) -> None:
"""Initialize with parameters."""
try:
import pymongo # noqa: F401
from pymongo import MongoClient # noqa: F401
except ImportError:
raise ImportError(
"`pymongo` package not found, please run `pip install pymongo`"
)
self.client: MongoClient = MongoClient(host, port)
self.max_docs = max_docs
def load_data(
self, db_name: str, collection_name: str, query_dict: Optional[Dict] = None
) -> List[Document]:
"""Load data from the input directory.
Args:
db_name (str): name of the database.
collection_name (str): name of the collection.
query_dict (Optional[Dict]): query to filter documents.
Defaults to None
Returns:
List[Document]: A list of documents.
"""
documents = []
db = self.client[db_name]
if query_dict is None:
cursor = db[collection_name].find()
else:
cursor = db[collection_name].find(query_dict)
for item in cursor:
if "text" not in item:
raise ValueError("`text` field not found in Mongo document.")
documents.append(Document(item["text"]))
return documents
|