-
Notifications
You must be signed in to change notification settings - Fork 2
/
MongoDBClass.py
executable file
·100 lines (79 loc) · 3.89 KB
/
MongoDBClass.py
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
89
90
91
92
93
94
95
96
97
98
99
# Author: PlanckBit
# MIT License
# Copyright (c) 2019 PlanckBit
import pymongo
from DatabaseClass import DatabaseClass
class MongoDBClass(DatabaseClass):
mongoDBInstanceCount = 0
def __init__(self, mongoDBServer: str):
DatabaseClass.__init__(self, "Mongo Database")
MongoDBClass.mongoDBInstanceCount += 1
self.mongoDBInstanceID = DatabaseClass.instanceSeedID
self.mongoDBServer = mongoDBServer
self.client = pymongo.MongoClient(self.mongoDBServer)
def mongoConnectDataBase(self, dataBaseName: str):
try:
self.dataBaseName = dataBaseName
return self.client[self.dataBaseName]
except pymongo.errors.ConnectionFailure:
print(e)
def mongoInsertOneRecord(self, collectionTable: str, dictRecord: dict):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.insert_one(dictRecord)
def mongoInsertManyRecords(self, collectionTable: str, listRecords: list):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
result = ""
try:
result = useCollectionTable.insert_many(listRecords)
except:
print(result)
def mongoDropDataBase(self, dataBaseName: str):
return self.client.drop_database(dataBaseName)
def mongoDropCollectionTable(self, collectionTable: str):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.drop()
def mongoFindOne(self, collectionTable: str):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.find_one()
def mongoFindAll(self, collectionTable: str, limit=0):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
dictNoId = {"_id": 0}
return useCollectionTable.find({}, dictNoId).limit(limit)
def mongoFindAllSpecificFields(self, collectionTable: str, dictFields: dict, limit=0):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.find({}, dictFields).limit(limit)
def mongoFindAllFilter(self, collectionTable: str, dictQuery: dict, limit=0):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.find(dictQuery).limit(limit)
def mongoFindAllSpecificFieldsFilter(self, collectionTable: str, dictQuery: dict, dictFields: dict, limit=0):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.find(dictQuery, dictFields).limit(limit)
def mongoDeleteOneRecord(self, collectionTable: str, dictQuery: dict):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.delete_one(dictQuery)
def mongoDeleteManyRecords(self, collectionTable: str, dictQuery: dict):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.delete_many(dictQuery)
def mongoUpdateOneRecord(self, collectionTable: str, dictQuery: dict, dictUpdate: dict):
useDB = self.client[self.dataBaseName]
useCollectionTable = useDB[collectionTable]
return useCollectionTable.update(dictQuery, dictUpdate)
def printMongoShowDatabases(self):
print(self.client.list_database_names())
def getMongoShowDatabases(self):
return self.client.list_database_names()
def getMongoDBClassInstanceCount(self):
return MongoDBClass.mongoDBInstanceCount
def __del__(self):
MongoDBClass.mongoDBInstanceCount -= 1
DatabaseClass.__del__(self, self.description, self.mongoDBInstanceID )