Skip to content

Instantly share code, notes, and snippets.

@NagariaHussain
Created November 27, 2022 07:20
  • Star 1 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save NagariaHussain/f72eab055fd13689e147c8b3d39878b6 to your computer and use it in GitHub Desktop.
# Copyright (c) 2022, Hussain and contributors
# For license information, please see license.txt
from frappe.model.document import Document
from pymongo import MongoClient
CONNECTION_STRING = "<YOUR-CONNECTION-STRING>"
def get_mongodb_client():
client = MongoClient(CONNECTION_STRING)
return client
def get_cars_collection():
client = get_mongodb_client()
db = client['cars-database']
cars = db.cars
return cars
class Car(Document):
def db_insert(self, *args, **kwargs):
cars = get_cars_collection()
d = self.get_valid_dict()
cars.insert_one(d)
def load_from_db(self):
cars = get_cars_collection()
d = cars.find_one({"name": self.name})
super(Document, self).__init__(d)
def db_update(self):
cars = get_cars_collection()
updated_values = self.get_valid_dict()
cars.update_one({"name": self.name}, {"$set": updated_values})
def delete(self):
cars = get_cars_collection()
cars.delete_one({"name": self.name})
@staticmethod
def get_list(args):
filters = args.filters
processed_filters = {}
if len(filters) > 0:
for filter in filters:
field = filter[1]
operator = filter[2]
value = filter[3]
operator = {
"=": "$eq",
">": "$gt",
"<": "$lt"
}[operator]
processed_filters[field] = {operator: value}
cars = get_cars_collection()
cars_list = []
for car in cars.find(processed_filters):
cars_list.append({
**car,
"_id": str(car["_id"]),
"_comment_count": 0
})
return cars_list
@staticmethod
def get_count(args):
return get_cars_collection().count_documents({})
@staticmethod
def get_stats(args):
return {}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment