osm_common/dbmongo.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Telefonica S.A.
   4 #
   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
   8 #
   9 #    http://www.apache.org/licenses/LICENSE-2.0
  10 #
  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
  14 # implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17
  18
  19 import logging
  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
  26 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  27
  28 # TODO consider use this decorator for database access retries
  29 # @retry_mongocall
  30 # def retry_mongocall(call):
  31 #     def _retry_mongocall(*args, **kwargs):
  32 #         retry = 1
  33 #         while True:
  34 #             try:
  35 #                 return call(*args, **kwargs)
  36 #             except pymongo.AutoReconnect as e:
  37 #                 if retry == 4:
  38 #                     raise DbException(e)
  39 #                 sleep(retry)
  40 #     return _retry_mongocall
  41
  42
  43 def deep_update(to_update, update_with):
  44     """
  45     Similar to deepcopy but recursively with nested dictionaries. 'to_update' dict is updated with a content copy of
  46     'update_with' dict recursively
  47     :param to_update: must be a dictionary to be modified
  48     :param update_with: must be a dictionary. It is not changed
  49     :return: to_update
  50     """
  51     for key in update_with:
  52         if key in to_update:
  53             if isinstance(to_update[key], dict) and isinstance(update_with[key], dict):
  54                 deep_update(to_update[key], update_with[key])
  55                 continue
  56         to_update[key] = deepcopy(update_with[key])
  57     return to_update
  58
  59
  60 class DbMongo(DbBase):
  61     conn_initial_timout = 120
  62     conn_timout = 10
  63
  64     def __init__(self, logger_name='db', master_password=None):
  65         super().__init__(logger_name, master_password)
  66         self.client = None
  67         self.db = None
  68
  69     def db_connect(self, config):
  70         """
  71         Connect to database
  72         :param config: Configuration of database
  73         :return: None or raises DbException on error
  74         """
  75         try:
  76             if "logger_name" in config:
  77                 self.logger = logging.getLogger(config["logger_name"])
  78             self.client = MongoClient(config["host"], config["port"])
  79             self.db = self.client[config["name"]]
  80             if "loglevel" in config:
  81                 self.logger.setLevel(getattr(logging, config['loglevel']))
  82             # get data to try a connection
  83             now = time()
  84             while True:
  85                 try:
  86                     self.db.users.find_one({"username": "admin"})
  87                     return
  88                 except errors.ConnectionFailure as e:
  89                     if time() - now >= self.conn_initial_timout:
  90                         raise
  91                     self.logger.info("Waiting to database up {}".format(e))
  92                     sleep(2)
  93         except errors.PyMongoError as e:
  94             raise DbException(e)
  95
  96     @staticmethod
  97     def _format_filter(q_filter):
  98         """
  99         Translate query string q_filter into mongo database filter
 100         :param q_filter: Query string content. Follows SOL005 section 4.3.2 guidelines, with the follow extensions and
 101         differences:
 102             It accept ".nq" (not equal) in addition to ".neq".
 103             For arrays you can specify index (concrete index must match), nothing (any index may match) or 'ANYINDEX'
 104             (two or more matches applies for the same array element). Examples:
 105                 with database register: {A: [{B: 1, C: 2}, {B: 6, C: 9}]}
 106                 query 'A.B=6' matches because array A contains one element with B equal to 6
 107                 query 'A.0.B=6' does no match because index 0 of array A contains B with value 1, but not 6
 108                 query 'A.B=6&A.C=2' matches because one element of array matches B=6 and other matchesC=2
 109                 query 'A.ANYINDEX.B=6&A.ANYINDEX.C=2' does not match because it is needed the same element of the
 110                     array matching both
 111
 112         Examples of translations from SOL005 to  >> mongo  # comment
 113             A=B; A.eq=B         >> A: B             # must contain key A and equal to B or be a list that contains B
 114             A.cont=B            >> A: B
 115             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
 116                 # B or C
 117             A.cont=B&A.cont=C; A.cont=B,C  >> A: {$in: [B, C]}
 118             A.ncont=B           >> A: {$nin: B}     # must not contain key A or if present not equal to B or if a list,
 119                 # it must not not contain B
 120             A.ncont=B,C; A.ncont=B&A.ncont=C    >> A: {$nin: [B,C]}     # must not contain key A or if present not equal
 121                 # neither B nor C; or if a list, it must not contain neither B nor C
 122             A.ne=B&A.ne=C; A.ne=B,C             >> A: {$nin: [B, C]}
 123             A.gt=B              >> A: {$gt: B}      # must contain key A and greater than B
 124             A.ne=B; A.neq=B         >> A: {$ne: B}          # must not contain key A or if present not equal to B, or if
 125                 # an array not contain B
 126             A.ANYINDEX.B=C          >> A: {$elemMatch: {B=C}
 127         :return: database mongo filter
 128         """
 129         try:
 130             db_filter = {}
 131             if not q_filter:
 132                 return db_filter
 133             for query_k, query_v in q_filter.items():
 134                 dot_index = query_k.rfind(".")
 135                 if dot_index > 1 and query_k[dot_index+1:] in ("eq", "ne", "gt", "gte", "lt", "lte", "cont",
 136                                                                "ncont", "neq"):
 137                     operator = "$" + query_k[dot_index + 1:]
 138                     if operator == "$neq":
 139                         operator = "$ne"
 140                     k = query_k[:dot_index]
 141                 else:
 142                     operator = "$eq"
 143                     k = query_k
 144
 145                 v = query_v
 146                 if isinstance(v, list):
 147                     if operator in ("$eq", "$cont"):
 148                         operator = "$in"
 149                         v = query_v
 150                     elif operator in ("$ne", "$ncont"):
 151                         operator = "$nin"
 152                         v = query_v
 153                     else:
 154                         v = query_v.join(",")
 155
 156                 if operator in ("$eq", "$cont"):
 157                     # v cannot be a comma separated list, because operator would have been changed to $in
 158                     db_v = v
 159                 elif operator == "$ncount":
 160                     # v cannot be a comma separated list, because operator would have been changed to $nin
 161                     db_v = {"$ne": v}
 162                 else:
 163                     db_v = {operator: v}
 164
 165                 # process the ANYINDEX word at k.
 166                 kleft, _, kright = k.rpartition(".ANYINDEX.")
 167                 while kleft:
 168                     k = kleft
 169                     db_v = {"$elemMatch": {kright: db_v}}
 170                     kleft, _, kright = k.rpartition(".ANYINDEX.")
 171
 172                 # insert in db_filter
 173                 # maybe db_filter[k] exist. e.g. in the query string for values between 5 and 8: "a.gt=5&a.lt=8"
 174                 deep_update(db_filter, {k: db_v})
 175
 176             return db_filter
 177         except Exception as e:
 178             raise DbException("Invalid query string filter at {}:{}. Error: {}".format(query_k, v, e),
 179                               http_code=HTTPStatus.BAD_REQUEST)
 180
 181     def get_list(self, table, q_filter=None):
 182         """
 183         Obtain a list of entries matching q_filter
 184         :param table: collection or table
 185         :param q_filter: Filter
 186         :return: a list (can be empty) with the found entries. Raises DbException on error
 187         """
 188         try:
 189             result = []
 190             collection = self.db[table]
 191             db_filter = self._format_filter(q_filter)
 192             rows = collection.find(db_filter)
 193             for row in rows:
 194                 result.append(row)
 195             return result
 196         except DbException:
 197             raise
 198         except Exception as e:  # TODO refine
 199             raise DbException(e)
 200
 201     def get_one(self, table, q_filter=None, fail_on_empty=True, fail_on_more=True):
 202         """
 203         Obtain one entry matching q_filter
 204         :param table: collection or table
 205         :param q_filter: Filter
 206         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 207         it raises a DbException
 208         :param fail_on_more: If more than one matches filter it returns one of then unless this flag is set tu True, so
 209         that it raises a DbException
 210         :return: The requested element, or None
 211         """
 212         try:
 213             db_filter = self._format_filter(q_filter)
 214             collection = self.db[table]
 215             if not (fail_on_empty and fail_on_more):
 216                 return collection.find_one(db_filter)
 217             rows = collection.find(db_filter)
 218             if rows.count() == 0:
 219                 if fail_on_empty:
 220                     raise DbException("Not found any {} with filter='{}'".format(table[:-1], q_filter),
 221                                       HTTPStatus.NOT_FOUND)
 222                 return None
 223             elif rows.count() > 1:
 224                 if fail_on_more:
 225                     raise DbException("Found more than one {} with filter='{}'".format(table[:-1], q_filter),
 226                                       HTTPStatus.CONFLICT)
 227             return rows[0]
 228         except Exception as e:  # TODO refine
 229             raise DbException(e)
 230
 231     def del_list(self, table, q_filter=None):
 232         """
 233         Deletes all entries that match q_filter
 234         :param table: collection or table
 235         :param q_filter: Filter
 236         :return: Dict with the number of entries deleted
 237         """
 238         try:
 239             collection = self.db[table]
 240             rows = collection.delete_many(self._format_filter(q_filter))
 241             return {"deleted": rows.deleted_count}
 242         except DbException:
 243             raise
 244         except Exception as e:  # TODO refine
 245             raise DbException(e)
 246
 247     def del_one(self, table, q_filter=None, fail_on_empty=True):
 248         """
 249         Deletes one entry that matches q_filter
 250         :param table: collection or table
 251         :param q_filter: Filter
 252         :param fail_on_empty: If nothing matches filter it returns '0' deleted unless this flag is set tu True, in
 253         which case it raises a DbException
 254         :return: Dict with the number of entries deleted
 255         """
 256         try:
 257             collection = self.db[table]
 258             rows = collection.delete_one(self._format_filter(q_filter))
 259             if rows.deleted_count == 0:
 260                 if fail_on_empty:
 261                     raise DbException("Not found any {} with filter='{}'".format(table[:-1], q_filter),
 262                                       HTTPStatus.NOT_FOUND)
 263                 return None
 264             return {"deleted": rows.deleted_count}
 265         except Exception as e:  # TODO refine
 266             raise DbException(e)
 267
 268     def create(self, table, indata):
 269         """
 270         Add a new entry at database
 271         :param table: collection or table
 272         :param indata: content to be added
 273         :return: database id of the inserted element. Raises a DbException on error
 274         """
 275         try:
 276             collection = self.db[table]
 277             data = collection.insert_one(indata)
 278             return data.inserted_id
 279         except Exception as e:  # TODO refine
 280             raise DbException(e)
 281
 282     def set_one(self, table, q_filter, update_dict, fail_on_empty=True):
 283         """
 284         Modifies an entry at database
 285         :param table: collection or table
 286         :param q_filter: Filter
 287         :param update_dict: Plain dictionary with the content to be updated. It is a dot separated keys and a value
 288         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 289         it raises a DbException
 290         :return: Dict with the number of entries modified. None if no matching is found.
 291         """
 292         try:
 293             collection = self.db[table]
 294             rows = collection.update_one(self._format_filter(q_filter), {"$set": update_dict})
 295             if rows.matched_count == 0:
 296                 if fail_on_empty:
 297                     raise DbException("Not found any {} with filter='{}'".format(table[:-1], q_filter),
 298                                       HTTPStatus.NOT_FOUND)
 299                 return None
 300             return {"modified": rows.modified_count}
 301         except Exception as e:  # TODO refine
 302             raise DbException(e)
 303
 304     def set_list(self, table, q_filter, update_dict):
 305         """
 306         Modifies al matching entries at database
 307         :param table: collection or table
 308         :param q_filter: Filter
 309         :param update_dict: Plain dictionary with the content to be updated. It is a dot separated keys and a value
 310         :return: Dict with the number of entries modified
 311         """
 312         try:
 313             collection = self.db[table]
 314             rows = collection.update_many(self._format_filter(q_filter), {"$set": update_dict})
 315             return {"modified": rows.modified_count}
 316         except Exception as e:  # TODO refine
 317             raise DbException(e)
 318
 319     def replace(self, table, _id, indata, fail_on_empty=True):
 320         """
 321         Replace the content of an entry
 322         :param table: collection or table
 323         :param _id: internal database id
 324         :param indata: content to replace
 325         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 326         it raises a DbException
 327         :return: Dict with the number of entries replaced
 328         """
 329         try:
 330             db_filter = {"_id": _id}
 331             collection = self.db[table]
 332             rows = collection.replace_one(db_filter, indata)
 333             if rows.matched_count == 0:
 334                 if fail_on_empty:
 335                     raise DbException("Not found any {} with _id='{}'".format(table[:-1], _id), HTTPStatus.NOT_FOUND)
 336                 return None
 337             return {"replaced": rows.modified_count}
 338         except Exception as e:  # TODO refine
 339             raise DbException(e)