1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Telefonica S.A.
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
20 from pymongo
import MongoClient
, errors
21 from osm_common
.dbbase
import DbException
, DbBase
22 from http
import HTTPStatus
23 from time
import time
, sleep
24 from copy
import deepcopy
25 from base64
import b64decode
27 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
29 # TODO consider use this decorator for database access retries
31 # def retry_mongocall(call):
32 # def _retry_mongocall(*args, **kwargs):
36 # return call(*args, **kwargs)
37 # except pymongo.AutoReconnect as e:
39 # raise DbException(e)
41 # return _retry_mongocall
44 def deep_update(to_update
, update_with
):
46 Similar to deepcopy but recursively with nested dictionaries. 'to_update' dict is updated with a content copy of
47 'update_with' dict recursively
48 :param to_update: must be a dictionary to be modified
49 :param update_with: must be a dictionary. It is not changed
52 for key
in update_with
:
54 if isinstance(to_update
[key
], dict) and isinstance(update_with
[key
], dict):
55 deep_update(to_update
[key
], update_with
[key
])
57 to_update
[key
] = deepcopy(update_with
[key
])
61 class DbMongo(DbBase
):
62 conn_initial_timout
= 120
65 def __init__(self
, logger_name
='db', lock
=False):
66 super().__init
__(logger_name
, lock
)
70 def db_connect(self
, config
, target_version
=None):
73 :param config: Configuration of database
74 :param target_version: if provided it checks if database contains required version, raising exception otherwise.
75 :return: None or raises DbException on error
78 if "logger_name" in config
:
79 self
.logger
= logging
.getLogger(config
["logger_name"])
80 master_key
= config
.get("commonkey") or config
.get("masterpassword")
82 self
.set_secret_key(master_key
)
84 self
.client
= MongoClient(config
["uri"])
86 self
.client
= MongoClient(config
["host"], config
["port"])
87 # TODO add as parameters also username=config.get("user"), password=config.get("password"))
88 # when all modules are ready
89 self
.db
= self
.client
[config
["name"]]
90 if "loglevel" in config
:
91 self
.logger
.setLevel(getattr(logging
, config
['loglevel']))
92 # get data to try a connection
96 version_data
= self
.get_one("admin", {"_id": "version"}, fail_on_empty
=False, fail_on_more
=True)
97 # check database status is ok
98 if version_data
and version_data
.get("status") != 'ENABLED':
99 raise DbException("Wrong database status '{}'".format(version_data
.get("status")),
100 http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
)
102 db_version
= None if not version_data
else version_data
.get("version")
103 if target_version
and target_version
!= db_version
:
104 raise DbException("Invalid database version {}. Expected {}".format(db_version
, target_version
))
106 if version_data
and version_data
.get("serial"):
107 self
.set_secret_key(b64decode(version_data
["serial"]))
108 self
.logger
.info("Connected to database {} version {}".format(config
["name"], db_version
))
110 except errors
.ConnectionFailure
as e
:
111 if time() - now
>= self
.conn_initial_timout
:
113 self
.logger
.info("Waiting to database up {}".format(e
))
115 except errors
.PyMongoError
as e
:
119 def _format_filter(q_filter
):
121 Translate query string q_filter into mongo database filter
122 :param q_filter: Query string content. Follows SOL005 section 4.3.2 guidelines, with the follow extensions and
124 It accept ".nq" (not equal) in addition to ".neq".
125 For arrays you can specify index (concrete index must match), nothing (any index may match) or 'ANYINDEX'
126 (two or more matches applies for the same array element). Examples:
127 with database register: {A: [{B: 1, C: 2}, {B: 6, C: 9}]}
128 query 'A.B=6' matches because array A contains one element with B equal to 6
129 query 'A.0.B=6' does no match because index 0 of array A contains B with value 1, but not 6
130 query 'A.B=6&A.C=2' matches because one element of array matches B=6 and other matchesC=2
131 query 'A.ANYINDEX.B=6&A.ANYINDEX.C=2' does not match because it is needed the same element of the
134 Examples of translations from SOL005 to >> mongo # comment
135 A=B; A.eq=B >> A: B # must contain key A and equal to B or be a list that contains B
137 A=B&A=C; A=B,C >> A: {$in: [B, C]} # must contain key A and equal to B or C or be a list that contains
139 A.cont=B&A.cont=C; A.cont=B,C >> A: {$in: [B, C]}
140 A.ncont=B >> A: {$nin: B} # must not contain key A or if present not equal to B or if a list,
141 # it must not not contain B
142 A.ncont=B,C; A.ncont=B&A.ncont=C >> A: {$nin: [B,C]} # must not contain key A or if present not equal
143 # neither B nor C; or if a list, it must not contain neither B nor C
144 A.ne=B&A.ne=C; A.ne=B,C >> A: {$nin: [B, C]}
145 A.gt=B >> A: {$gt: B} # must contain key A and greater than B
146 A.ne=B; A.neq=B >> A: {$ne: B} # must not contain key A or if present not equal to B, or if
147 # an array not contain B
148 A.ANYINDEX.B=C >> A: {$elemMatch: {B=C}
149 :return: database mongo filter
155 for query_k
, query_v
in q_filter
.items():
156 dot_index
= query_k
.rfind(".")
157 if dot_index
> 1 and query_k
[dot_index
+1:] in ("eq", "ne", "gt", "gte", "lt", "lte", "cont",
159 operator
= "$" + query_k
[dot_index
+ 1:]
160 if operator
== "$neq":
162 k
= query_k
[:dot_index
]
168 if isinstance(v
, list):
169 if operator
in ("$eq", "$cont"):
172 elif operator
in ("$ne", "$ncont"):
176 v
= query_v
.join(",")
178 if operator
in ("$eq", "$cont"):
179 # v cannot be a comma separated list, because operator would have been changed to $in
181 elif operator
== "$ncount":
182 # v cannot be a comma separated list, because operator would have been changed to $nin
187 # process the ANYINDEX word at k.
188 kleft
, _
, kright
= k
.rpartition(".ANYINDEX.")
191 db_v
= {"$elemMatch": {kright
: db_v
}}
192 kleft
, _
, kright
= k
.rpartition(".ANYINDEX.")
194 # insert in db_filter
195 # maybe db_filter[k] exist. e.g. in the query string for values between 5 and 8: "a.gt=5&a.lt=8"
196 deep_update(db_filter
, {k
: db_v
})
199 except Exception as e
:
200 raise DbException("Invalid query string filter at {}:{}. Error: {}".format(query_k
, v
, e
),
201 http_code
=HTTPStatus
.BAD_REQUEST
)
203 def get_list(self
, table
, q_filter
=None):
205 Obtain a list of entries matching q_filter
206 :param table: collection or table
207 :param q_filter: Filter
208 :return: a list (can be empty) with the found entries. Raises DbException on error
213 collection
= self
.db
[table
]
214 db_filter
= self
._format
_filter
(q_filter
)
215 rows
= collection
.find(db_filter
)
221 except Exception as e
: # TODO refine
224 def get_one(self
, table
, q_filter
=None, fail_on_empty
=True, fail_on_more
=True):
226 Obtain one entry matching q_filter
227 :param table: collection or table
228 :param q_filter: Filter
229 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
230 it raises a DbException
231 :param fail_on_more: If more than one matches filter it returns one of then unless this flag is set tu True, so
232 that it raises a DbException
233 :return: The requested element, or None
236 db_filter
= self
._format
_filter
(q_filter
)
238 collection
= self
.db
[table
]
239 if not (fail_on_empty
and fail_on_more
):
240 return collection
.find_one(db_filter
)
241 rows
= collection
.find(db_filter
)
242 if rows
.count() == 0:
244 raise DbException("Not found any {} with filter='{}'".format(table
[:-1], q_filter
),
245 HTTPStatus
.NOT_FOUND
)
247 elif rows
.count() > 1:
249 raise DbException("Found more than one {} with filter='{}'".format(table
[:-1], q_filter
),
252 except Exception as e
: # TODO refine
255 def del_list(self
, table
, q_filter
=None):
257 Deletes all entries that match q_filter
258 :param table: collection or table
259 :param q_filter: Filter
260 :return: Dict with the number of entries deleted
264 collection
= self
.db
[table
]
265 rows
= collection
.delete_many(self
._format
_filter
(q_filter
))
266 return {"deleted": rows
.deleted_count
}
269 except Exception as e
: # TODO refine
272 def del_one(self
, table
, q_filter
=None, fail_on_empty
=True):
274 Deletes one entry that matches q_filter
275 :param table: collection or table
276 :param q_filter: Filter
277 :param fail_on_empty: If nothing matches filter it returns '0' deleted unless this flag is set tu True, in
278 which case it raises a DbException
279 :return: Dict with the number of entries deleted
283 collection
= self
.db
[table
]
284 rows
= collection
.delete_one(self
._format
_filter
(q_filter
))
285 if rows
.deleted_count
== 0:
287 raise DbException("Not found any {} with filter='{}'".format(table
[:-1], q_filter
),
288 HTTPStatus
.NOT_FOUND
)
290 return {"deleted": rows
.deleted_count
}
291 except Exception as e
: # TODO refine
294 def create(self
, table
, indata
):
296 Add a new entry at database
297 :param table: collection or table
298 :param indata: content to be added
299 :return: database id of the inserted element. Raises a DbException on error
303 collection
= self
.db
[table
]
304 data
= collection
.insert_one(indata
)
305 return data
.inserted_id
306 except Exception as e
: # TODO refine
309 def set_one(self
, table
, q_filter
, update_dict
, fail_on_empty
=True, unset
=None, pull
=None, push
=None):
311 Modifies an entry at database
312 :param table: collection or table
313 :param q_filter: Filter
314 :param update_dict: Plain dictionary with the content to be updated. It is a dot separated keys and a value
315 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
316 it raises a DbException
317 :param unset: Plain dictionary with the content to be removed if exist. It is a dot separated keys, value is
318 ignored. If not exist, it is ignored
319 :param pull: Plain dictionary with the content to be removed from an array. It is a dot separated keys and value
320 if exist in the array is removed. If not exist, it is ignored
321 :param push: Plain dictionary with the content to be appended to an array. It is a dot separated keys and value
322 is appended to the end of the array
323 :return: Dict with the number of entries modified. None if no matching is found.
328 db_oper
["$set"] = update_dict
330 db_oper
["$unset"] = unset
332 db_oper
["$pull"] = pull
334 db_oper
["$push"] = push
337 collection
= self
.db
[table
]
338 rows
= collection
.update_one(self
._format
_filter
(q_filter
), db_oper
)
339 if rows
.matched_count
== 0:
341 raise DbException("Not found any {} with filter='{}'".format(table
[:-1], q_filter
),
342 HTTPStatus
.NOT_FOUND
)
344 return {"modified": rows
.modified_count
}
345 except Exception as e
: # TODO refine
348 def set_list(self
, table
, q_filter
, update_dict
):
350 Modifies al matching entries at database
351 :param table: collection or table
352 :param q_filter: Filter
353 :param update_dict: Plain dictionary with the content to be updated. It is a dot separated keys and a value
354 :return: Dict with the number of entries modified
358 collection
= self
.db
[table
]
359 rows
= collection
.update_many(self
._format
_filter
(q_filter
), {"$set": update_dict
})
360 return {"modified": rows
.modified_count
}
361 except Exception as e
: # TODO refine
364 def replace(self
, table
, _id
, indata
, fail_on_empty
=True):
366 Replace the content of an entry
367 :param table: collection or table
368 :param _id: internal database id
369 :param indata: content to replace
370 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
371 it raises a DbException
372 :return: Dict with the number of entries replaced
375 db_filter
= {"_id": _id
}
377 collection
= self
.db
[table
]
378 rows
= collection
.replace_one(db_filter
, indata
)
379 if rows
.matched_count
== 0:
381 raise DbException("Not found any {} with _id='{}'".format(table
[:-1], _id
), HTTPStatus
.NOT_FOUND
)
383 return {"replaced": rows
.modified_count
}
384 except Exception as e
: # TODO refine