lcm/osm_common/dbmemory.py

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