Skip to content

Instantly share code, notes, and snippets.

@NagariaHussain
Created November 28, 2022 03:43
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save NagariaHussain/d8ee43f9ef095eb4a6d9197f1175219b to your computer and use it in GitHub Desktop.
# Copyright (c) 2022, Hussain and contributors
# For license information, please see license.txt
from pymongo import MongoClient
from abc import ABC, abstractstaticmethod
from frappe.model.document import Document
class MongoDBDocument(Document, ABC):
@abstractstaticmethod
def get_collection():
raise NotImplementedError
def db_insert(self, *args, **kwargs):
collection = self.get_collection()
d = self.get_valid_dict()
collection.insert_one(d)
def load_from_db(self):
collection = self.get_collection()
d = collection.find_one({"name": self.name})
super(Document, self).__init__(d)
def db_update(self):
collection = self.get_collection()
updated_values = self.get_valid_dict()
collection.update_one({"name": self.name}, {"$set": updated_values})
def delete(self):
collection = self.get_collection()
collection.delete_one({"name": self.name})
@classmethod
def get_list(cls, args):
collection = cls.get_collection()
collection_list = []
for doc in collection.find():
collection_list.append({
**doc,
"_id": str(doc["_id"]),
"_comment_count": 0
})
return collection_list
@classmethod
def get_count(cls, args):
return cls.get_collection().count_documents({})
@staticmethod
def get_stats(args):
return {}
CONNECTION_STRING = "<your_connection_string>"
def get_mongodb_client():
return MongoClient(CONNECTION_STRING)
class Car(MongoDBDocument):
@staticmethod
def get_collection():
client = get_mongodb_client()
db = client['cars-database']
cars = db.cars
return cars
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment