Spaces:
Sleeping
Sleeping
File size: 2,920 Bytes
172e21d 75115cd |
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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
import os
from pymongo import MongoClient
from dotenv import load_dotenv
from typing import List, Dict, Optional
from application.utils.logger import get_logger
logger = get_logger()
load_dotenv()
DB_NAME = "sustainability_reports_db"
def get_mongo_client():
try:
client = MongoClient(os.getenv("MONGODB_URI"))
return client
except Exception as e:
logger.exception(f"An unexpected error occurred while connecting to MongoDB: {str(e)}")
return None
def store_document(collection_name: str, document: Dict) -> Optional[str]:
"""
Stores a document in MongoDB if it doesn't already exist.
Args:
collection_name (str): Name of the MongoDB collection.
document (Dict): The document to be inserted.
Returns:
Optional[str]: Inserted document ID if successful, None otherwise.
"""
try:
client = get_mongo_client()
if client is None:
logger.error("MongoDB client is not available.")
return None
db = client.get_database(DB_NAME)
collection = db[collection_name]
# Check if a similar document already exists
existing_document = collection.find_one(document)
if existing_document:
logger.info(f"Document already exists with ID: {existing_document['_id']}")
return str(existing_document['_id'])
# If no existing document, insert the new one
result = collection.insert_one(document)
logger.info(f"New document inserted with ID: {result.inserted_id}")
return str(result.inserted_id)
except Exception as e:
logger.exception(f"An unexpected error occurred: {str(e)}")
return None
def retrieve_documents(collection_name: str, query: Optional[Dict] = None) -> List[Dict]:
"""
Retrieves documents from the specified MongoDB collection.
Args:
collection_name (str): Name of the MongoDB collection.
query (Optional[Dict]): A MongoDB query filter. Defaults to {} (fetch all documents).
Returns:
List[Dict]: A list of documents matching the query. Empty list if none found or error occurs.
"""
try:
client = get_mongo_client()
if client is None:
logger.error("MongoDB client is not available.")
return []
db = client.get_database(DB_NAME)
collection = db[collection_name]
documents_cursor = collection.find(query or {})
documents = list(documents_cursor)
logger.info(f"Retrieved {len(documents)} documents from collection: {collection_name}")
return documents
except Exception as e:
logger.exception(f"An error occurred while retrieving documents: {str(e)}")
return []
# all_docs = retrieve_documents("Zalando")
# print(all_docs) |