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