osm_common/dbmemory.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 import logging
  19 from osm_common.dbbase import DbException, DbBase
  20 from http import HTTPStatus
  21 from uuid import uuid4
  22 from copy import deepcopy
  23
  24 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  25
  26
  27 class DbMemory(DbBase):
  28
  29     def __init__(self, logger_name='db', master_password=None):
  30         super().__init__(logger_name, master_password)
  31         self.db = {}
  32
  33     def db_connect(self, config):
  34         """
  35         Connect to database
  36         :param config: Configuration of database
  37         :return: None or raises DbException on error
  38         """
  39         if "logger_name" in config:
  40             self.logger = logging.getLogger(config["logger_name"])
  41
  42     @staticmethod
  43     def _format_filter(q_filter):
  44         return q_filter    # TODO
  45
  46     def _find(self, table, q_filter):
  47         for i, row in enumerate(self.db.get(table, ())):
  48             match = True
  49             if q_filter:
  50                 for k, v in q_filter.items():
  51                     if k not in row or v != row[k]:
  52                         match = False
  53             if match:
  54                 yield i, row
  55
  56     def get_list(self, table, q_filter=None):
  57         """
  58         Obtain a list of entries matching q_filter
  59         :param table: collection or table
  60         :param q_filter: Filter
  61         :return: a list (can be empty) with the found entries. Raises DbException on error
  62         """
  63         try:
  64             result = []
  65             for _, row in self._find(table, self._format_filter(q_filter)):
  66                 result.append(deepcopy(row))
  67             return result
  68         except DbException:
  69             raise
  70         except Exception as e:  # TODO refine
  71             raise DbException(str(e))
  72
  73     def get_one(self, table, q_filter=None, fail_on_empty=True, fail_on_more=True):
  74         """
  75         Obtain one entry matching q_filter
  76         :param table: collection or table
  77         :param q_filter: Filter
  78         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
  79         it raises a DbException
  80         :param fail_on_more: If more than one matches filter it returns one of then unless this flag is set tu True, so
  81         that it raises a DbException
  82         :return: The requested element, or None
  83         """
  84         try:
  85             result = None
  86             for _, row in self._find(table, self._format_filter(q_filter)):
  87                 if not fail_on_more:
  88                     return deepcopy(row)
  89                 if result:
  90                     raise DbException("Found more than one entry with filter='{}'".format(q_filter),
  91                                       HTTPStatus.CONFLICT.value)
  92                 result = row
  93             if not result and fail_on_empty:
  94                 raise DbException("Not found entry with filter='{}'".format(q_filter), HTTPStatus.NOT_FOUND)
  95             return deepcopy(result)
  96         except Exception as e:  # TODO refine
  97             raise DbException(str(e))
  98
  99     def del_list(self, table, q_filter=None):
 100         """
 101         Deletes all entries that match q_filter
 102         :param table: collection or table
 103         :param q_filter: Filter
 104         :return: Dict with the number of entries deleted
 105         """
 106         try:
 107             id_list = []
 108             for i, _ in self._find(table, self._format_filter(q_filter)):
 109                 id_list.append(i)
 110             deleted = len(id_list)
 111             for i in reversed(id_list):
 112                 del self.db[table][i]
 113             return {"deleted": deleted}
 114         except DbException:
 115             raise
 116         except Exception as e:  # TODO refine
 117             raise DbException(str(e))
 118
 119     def del_one(self, table, q_filter=None, fail_on_empty=True):
 120         """
 121         Deletes one entry that matches q_filter
 122         :param table: collection or table
 123         :param q_filter: Filter
 124         :param fail_on_empty: If nothing matches filter it returns '0' deleted unless this flag is set tu True, in
 125         which case it raises a DbException
 126         :return: Dict with the number of entries deleted
 127         """
 128         try:
 129             for i, _ in self._find(table, self._format_filter(q_filter)):
 130                 break
 131             else:
 132                 if fail_on_empty:
 133                     raise DbException("Not found entry with filter='{}'".format(q_filter), HTTPStatus.NOT_FOUND)
 134                 return None
 135             del self.db[table][i]
 136             return {"deleted": 1}
 137         except Exception as e:  # TODO refine
 138             raise DbException(str(e))
 139
 140     def replace(self, table, _id, indata, fail_on_empty=True):
 141         """
 142         Replace the content of an entry
 143         :param table: collection or table
 144         :param _id: internal database id
 145         :param indata: content to replace
 146         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 147         it raises a DbException
 148         :return: Dict with the number of entries replaced
 149         """
 150         try:
 151             for i, _ in self._find(table, self._format_filter({"_id": _id})):
 152                 break
 153             else:
 154                 if fail_on_empty:
 155                     raise DbException("Not found entry with _id='{}'".format(_id), HTTPStatus.NOT_FOUND)
 156                 return None
 157             self.db[table][i] = deepcopy(indata)
 158             return {"updated": 1}
 159         except DbException:
 160             raise
 161         except Exception as e:  # TODO refine
 162             raise DbException(str(e))
 163
 164     def create(self, table, indata):
 165         """
 166         Add a new entry at database
 167         :param table: collection or table
 168         :param indata: content to be added
 169         :return: database id of the inserted element. Raises a DbException on error
 170         """
 171         try:
 172             id = indata.get("_id")
 173             if not id:
 174                 id = str(uuid4())
 175                 indata["_id"] = id
 176             if table not in self.db:
 177                 self.db[table] = []
 178             self.db[table].append(deepcopy(indata))
 179             return id
 180         except Exception as e:  # TODO refine
 181             raise DbException(str(e))
 182
 183
 184 if __name__ == '__main__':
 185     # some test code
 186     db = DbMemory()
 187     db.create("test", {"_id": 1, "data": 1})
 188     db.create("test", {"_id": 2, "data": 2})
 189     db.create("test", {"_id": 3, "data": 3})
 190     print("must be 3 items:", db.get_list("test"))
 191     print("must return item 2:", db.get_list("test", {"_id": 2}))
 192     db.del_one("test", {"_id": 2})
 193     print("must be emtpy:", db.get_list("test", {"_id": 2}))