Spaces:
Runtime error
Runtime error
File size: 2,256 Bytes
b699122 |
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 61 62 63 64 65 66 67 68 69 70 71 72 73 |
"""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: Optional[str] = None,
port: Optional[int] = None,
uri: Optional[str] = None,
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`"
)
if uri:
if uri is None:
raise ValueError("Either `host` and `port` or `uri` must be provided.")
self.client: MongoClient = MongoClient(uri)
else:
if host is None or port is None:
raise ValueError("Either `host` and `port` or `uri` must be provided.")
self.client = 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
|