osm_nbi/dbmemory.py

   1 import logging
   2 from dbbase import DbException, DbBase
   3 from http import HTTPStatus
   4 from uuid import uuid4
   5 from copy import deepcopy
   6
   7 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
   8
   9
  10 class DbMemory(DbBase):
  11
  12     def __init__(self, logger_name='db'):
  13         self.logger = logging.getLogger(logger_name)
  14         self.db = {}
  15
  16     def db_connect(self, config):
  17         if "logger_name" in config:
  18             self.logger = logging.getLogger(config["logger_name"])
  19
  20     @staticmethod
  21     def _format_filter(filter):
  22         return filter    # TODO
  23
  24     def _find(self, table, filter):
  25         for i, row in enumerate(self.db.get(table, ())):
  26             match = True
  27             if filter:
  28                 for k, v in filter.items():
  29                     if k not in row or v != row[k]:
  30                         match = False
  31             if match:
  32                 yield i, row
  33
  34     def get_list(self, table, filter={}):
  35         try:
  36             l = []
  37             for _, row in self._find(table, self._format_filter(filter)):
  38                 l.append(deepcopy(row))
  39             return l
  40         except DbException:
  41             raise
  42         except Exception as e:  # TODO refine
  43             raise DbException(str(e))
  44
  45     def get_one(self, table, filter={}, fail_on_empty=True, fail_on_more=True):
  46         try:
  47             l = None
  48             for _, row in self._find(table, self._format_filter(filter)):
  49                 if not fail_on_more:
  50                     return deepcopy(row)
  51                 if l:
  52                     raise DbException("Found more than one entry with filter='{}'".format(filter),
  53                                       HTTPStatus.CONFLICT.value)
  54                 l = row
  55             if not l and fail_on_empty:
  56                 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
  57             return deepcopy(l)
  58         except Exception as e:  # TODO refine
  59             raise DbException(str(e))
  60
  61     def del_list(self, table, filter={}):
  62         try:
  63             id_list = []
  64             for i, _ in self._find(table, self._format_filter(filter)):
  65                 id_list.append(i)
  66             deleted = len(id_list)
  67             for i in id_list:
  68                 del self.db[table][i]
  69             return {"deleted": deleted}
  70         except DbException:
  71             raise
  72         except Exception as e:  # TODO refine
  73             raise DbException(str(e))
  74
  75     def del_one(self, table, filter={}, fail_on_empty=True):
  76         try:
  77             for i, _ in self._find(table, self._format_filter(filter)):
  78                 break
  79             else:
  80                 if fail_on_empty:
  81                     raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
  82                 return None
  83             del self.db[table][i]
  84             return {"deleted": 1}
  85         except Exception as e:  # TODO refine
  86             raise DbException(str(e))
  87
  88     def replace(self, table, filter, indata, fail_on_empty=True):
  89         try:
  90             for i, _ in self._find(table, self._format_filter(filter)):
  91                 break
  92             else:
  93                 if fail_on_empty:
  94                     raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
  95                 return None
  96             self.db[table][i] = deepcopy(indata)
  97             return {"upadted": 1}
  98         except Exception as e:  # TODO refine
  99             raise DbException(str(e))
 100
 101     def create(self, table, indata):
 102         try:
 103             id = indata.get("_id")
 104             if not id:
 105                 id = str(uuid4())
 106                 indata["_id"] = id
 107             if table not in self.db:
 108                 self.db[table] = []
 109             self.db[table].append(deepcopy(indata))
 110             return id
 111         except Exception as e:  # TODO refine
 112             raise DbException(str(e))
 113
 114
 115 if __name__ == '__main__':
 116     # some test code
 117     db = dbmemory()
 118     db.create("test", {"_id": 1, "data": 1})
 119     db.create("test", {"_id": 2, "data": 2})
 120     db.create("test", {"_id": 3, "data": 3})
 121     print("must be 3 items:", db.get_list("test"))
 122     print("must return item 2:", db.get_list("test", {"_id": 2}))
 123     db.del_one("test", {"_id": 2})
 124     print("must be emtpy:", db.get_list("test", {"_id": 2}))