-
Notifications
You must be signed in to change notification settings - Fork 2
/
firebase.py
207 lines (171 loc) · 7.67 KB
/
firebase.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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
import firebase_admin
from firebase_admin import credentials, firestore, storage
class Firebase:
def __init__(self):
"""
Initializes the Firebase instance.
This constructor checks if the Firebase app has already been initialized and
either initializes a new app or uses an existing one.
Raises:
Exception: If unable to initialize Firebase.
"""
try:
self.cred = credentials.Certificate('firebase-creds.json')
except Exception as e:
self.cred = credentials.Certificate('firebase-creds.json')
# Check if the app has already been initialized
if not firebase_admin._apps:
self.app = firebase_admin.initialize_app(self.cred, {
'storageBucket': "inf2003-e21a8.appspot.com"
})
else:
# If the app is already initialized, use the existing app
self.app = firebase_admin._apps
self.db = firestore.client()
self.bucket = storage.bucket()
def close_db(self):
"""
Closes the Firestore database connection.
"""
self.db.close()
def get(self, collection_path, document_id=None, query=None, queryType="and"):
"""
Fetches data from the Firestore database.
Args:
collection_path (str): The path to the Firestore collection.
document_id (str, optional): The ID of the specific document to fetch.
query (list, optional): A list of query filters to apply when fetching documents.
queryType (str): The type of query to perform ('and' or 'or') when multiple filters are provided.
Returns:
dict or list: The fetched data as a dictionary (for single document) or a list of dictionaries (for multiple documents).
Raises:
Exception: If an error occurs during database access.
"""
try:
collection_ref = self.db.collection(collection_path)
if document_id:
# Case 2: Fetch a specific document
doc_ref = collection_ref.document(document_id)
doc = doc_ref.get()
if doc.exists:
return doc.to_dict()
else:
raise Exception("Document not found")
elif query:
# Case 3: Fetch documents based on a query
# If there is a query, we need to loop through each of the queries
filters = []
for filter in query:
field, operator, value = filter
filters.append(firestore.AsyncFieldFilter(field, operator, value))
if queryType == "or":
final_query = firestore.AsyncOr(filters=filters)
else:
final_query = firestore.AsyncAnd(filters=filters)
collection_ref = collection_ref.where(filter=final_query)
documents = collection_ref.stream()
result = [doc.to_dict() for doc in documents]
return result
else:
# Case 1: Fetch all data from the collection
documents = collection_ref.stream()
result = [doc.to_dict() for doc in documents]
return result
except Exception as e:
# Handle exceptions and return appropriate error messages
return str(e)
def add(self, collection_path, data, document_id=None):
"""
Adds a document to the Firestore database.
Args:
collection_path (str): The path to the Firestore collection.
data (dict): The data to be added as a dictionary.
document_id (str, optional): The ID of the specific document to add.
Returns:
None
"""
collection_ref = self.db.collection(collection_path)
if document_id:
# Case 1: Add a document with a specific ID
doc_ref = collection_ref.document(document_id)
doc_ref.set(data)
else:
# Case 2: Add a document with a random ID
collection_ref.add(data)
def update(self, collection_path, document_id, data):
"""
Updates a document in the Firestore database.
Args:
collection_path (str): The path to the Firestore collection.
document_id (str): The ID of the specific document to update.
data (dict): The data to be updated as a dictionary.
Returns:
None
"""
collection_ref = self.db.collection(collection_path)
doc_ref = collection_ref.document(document_id)
doc_ref.update(data)
def delete(self, collection_path, document_id=None, query=None, queryType="and"):
"""
Deletes documents from the Firestore database.
Args:
collection_path (str): The path to the Firestore collection.
document_id (str, optional): The ID of the specific document to delete.
query (list, optional): A list of query filters to apply when deleting documents.
Returns:
None
"""
collection_ref = self.db.collection(collection_path)
if document_id:
# Case 1: Delete a specific document
doc_ref = collection_ref.document(document_id)
doc_ref.delete()
elif query:
# Case 2: Delete all documents that match the query
# If there is a query, we need to loop through each of the queries
filters = []
for filter in query:
field, operator, value = filter
filters.append(firestore.FieldFilter(field, operator, value))
if queryType == "or":
final_query = firestore.Or(filters=filters)
else:
final_query = firestore.And(filters=filters)
collection_ref = collection_ref.where(filter=final_query)
documents = collection_ref.stream()
for doc in documents:
doc.reference.delete()
else:
# Case 3: Delete all documents in the collection
documents = collection_ref.stream()
for doc in documents:
doc.reference.delete()
def increment(self, collection_path, document_id=None, query=None, field=None, value=1):
"""
Increments fields in documents in the Firestore database.
Args:
collection_path (str): The path to the Firestore collection.
document_id (str, optional): The ID of the specific document to update.
query (list, optional): A list of query filters to apply when updating documents.
field (str): The field to increment.
value (int): The value by which to increment the field.
Returns:
None
"""
collection_ref = self.db.collection(collection_path)
if query:
# Case 1: Increment all documents that match the query
# If there is a query, we need to loop through each of the queries
filters = []
for filter in query:
field, operator, value = filter
filters.append(firestore.FieldFilter(field, operator, value))
final_query = firestore.And(filters=filters)
collection_ref = collection_ref.where(filter=final_query)
documents = collection_ref.stream()
for doc in documents:
doc.reference.update({field: firestore.Increment(value)})
else:
# Case 2: Increment a specific document
doc_ref = collection_ref.document(document_id)
doc_ref.update({field: firestore.Increment(value)})