lcm/osm_common/dbmongo.py

   1 #import pymongo
   2 import logging
   3 from pymongo import MongoClient, errors
   4 from dbbase import DbException, DbBase
   5 from http import HTTPStatus
   6 from time import time, sleep
   7
   8 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
   9
  10
  11 class DbMongo(DbBase):
  12     conn_initial_timout = 120
  13     conn_timout = 10
  14
  15     def __init__(self, logger_name='db'):
  16         self.logger = logging.getLogger(logger_name)
  17
  18     def db_connect(self, config):
  19         try:
  20             if "logger_name" in config:
  21                 self.logger = logging.getLogger(config["logger_name"])
  22             self.client = MongoClient(config["host"], config["port"])
  23             self.db = self.client[config["name"]]
  24             if "loglevel" in config:
  25                 self.logger.setLevel(getattr(logging, config['loglevel']))
  26             # get data to try a connection
  27             now = time()
  28             while True:
  29                 try:
  30                     self.db.users.find_one({"username": "admin"})
  31                     return
  32                 except errors.ConnectionFailure as e:
  33                     if time() - now >= self.conn_initial_timout:
  34                         raise
  35                     self.logger.info("Waiting to database up {}".format(e))
  36                     sleep(2)
  37         except errors.PyMongoError as e:
  38             raise DbException(str(e))
  39
  40     def db_disconnect(self):
  41         pass  # TODO
  42
  43     @staticmethod
  44     def _format_filter(filter):
  45         try:
  46             db_filter = {}
  47             for query_k, query_v in filter.items():
  48                 dot_index = query_k.rfind(".")
  49                 if dot_index > 1 and query_k[dot_index+1:] in ("eq", "ne", "gt", "gte", "lt", "lte", "cont",
  50                                                                "ncont", "neq"):
  51                     operator = "$" + query_k[dot_index+1:]
  52                     if operator == "$neq":
  53                         operator = "$nq"
  54                     k = query_k[:dot_index]
  55                 else:
  56                     operator = "$eq"
  57                     k = query_k
  58
  59                 v = query_v
  60                 if isinstance(v, list):
  61                     if operator in ("$eq", "$cont"):
  62                         operator = "$in"
  63                         v = query_v
  64                     elif operator in ("$ne", "$ncont"):
  65                         operator = "$nin"
  66                         v = query_v
  67                     else:
  68                         v = query_v.join(",")
  69
  70                 if operator in ("$eq", "$cont"):
  71                     # v cannot be a comma separated list, because operator would have been changed to $in
  72                     db_filter[k] = v
  73                 elif operator == "$ncount":
  74                     # v cannot be a comma separated list, because operator would have been changed to $nin
  75                     db_filter[k] = {"$ne": v}
  76                 else:
  77                     # maybe db_filter[k] exist. e.g. in the query string for values between 5 and 8: "a.gt=5&a.lt=8"
  78                     if k not in db_filter:
  79                         db_filter[k] = {}
  80                     db_filter[k][operator] = v
  81
  82             return db_filter
  83         except Exception as e:
  84             raise DbException("Invalid query string filter at {}:{}. Error: {}".format(query_k, v, e),
  85                               http_code=HTTPStatus.BAD_REQUEST)
  86
  87
  88     def get_list(self, table, filter={}):
  89         try:
  90             l = []
  91             collection = self.db[table]
  92             rows = collection.find(self._format_filter(filter))
  93             for row in rows:
  94                 l.append(row)
  95             return l
  96         except DbException:
  97             raise
  98         except Exception as e:  # TODO refine
  99             raise DbException(str(e))
 100
 101     def get_one(self, table, filter={}, fail_on_empty=True, fail_on_more=True):
 102         try:
 103             if filter:
 104                 filter = self._format_filter(filter)
 105             collection = self.db[table]
 106             if not (fail_on_empty and fail_on_more):
 107                 return collection.find_one(filter)
 108             rows = collection.find(filter)
 109             if rows.count() == 0:
 110                 if fail_on_empty:
 111                     raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
 112                 return None
 113             elif rows.count() > 1:
 114                 if fail_on_more:
 115                     raise DbException("Found more than one entry with filter='{}'".format(filter),
 116                                       HTTPStatus.CONFLICT)
 117             return rows[0]
 118         except Exception as e:  # TODO refine
 119             raise DbException(str(e))
 120
 121     def del_list(self, table, filter={}):
 122         try:
 123             collection = self.db[table]
 124             rows = collection.delete_many(self._format_filter(filter))
 125             return {"deleted": rows.deleted_count}
 126         except DbException:
 127             raise
 128         except Exception as e:  # TODO refine
 129             raise DbException(str(e))
 130
 131     def del_one(self, table, filter={}, fail_on_empty=True):
 132         try:
 133             collection = self.db[table]
 134             rows = collection.delete_one(self._format_filter(filter))
 135             if rows.deleted_count == 0:
 136                 if fail_on_empty:
 137                     raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
 138                 return None
 139             return {"deleted": rows.deleted_count}
 140         except Exception as e:  # TODO refine
 141             raise DbException(str(e))
 142
 143     def create(self, table, indata):
 144         try:
 145             collection = self.db[table]
 146             data = collection.insert_one(indata)
 147             return data.inserted_id
 148         except Exception as e:  # TODO refine
 149             raise DbException(str(e))
 150
 151     def set_one(self, table, filter, update_dict, fail_on_empty=True):
 152         try:
 153             collection = self.db[table]
 154             rows = collection.update_one(self._format_filter(filter), {"$set": update_dict})
 155             if rows.updated_count == 0:
 156                 if fail_on_empty:
 157                     raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
 158                 return None
 159             return {"deleted": rows.deleted_count}
 160         except Exception as e:  # TODO refine
 161             raise DbException(str(e))
 162
 163     def replace(self, table, id, indata, fail_on_empty=True):
 164         try:
 165             collection = self.db[table]
 166             rows = collection.replace_one({"_id": id}, indata)
 167             if rows.modified_count == 0:
 168                 if fail_on_empty:
 169                     raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
 170                 return None
 171             return {"replace": rows.modified_count}
 172         except Exception as e:  # TODO refine
 173             raise DbException(str(e))