1 from dbbase
import DbException
, dbbase
2 from http
import HTTPStatus
4 from copy
import deepcopy
7 class dbmemory(dbbase
):
13 def _format_filter(filter):
16 def _find(self
, table
, filter):
17 for i
, row
in enumerate(self
.db
.get(table
, ())):
20 for k
, v
in filter.items():
21 if k
not in row
or v
!= row
[k
]:
26 def get_list(self
, table
, filter={}):
29 for _
, row
in self
._find
(table
, self
._format
_filter
(filter)):
30 l
.append(deepcopy(row
))
34 except Exception as e
: # TODO refine
35 raise DbException(str(e
))
37 def get_one(self
, table
, filter={}, fail_on_empty
=True, fail_on_more
=True):
40 for _
, row
in self
._find
(table
, self
._format
_filter
(filter)):
44 raise DbException("Found more than one entry with filter='{}'".format(filter),
45 HTTPStatus
.CONFLICT
.value
)
47 if not l
and fail_on_empty
:
48 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus
.NOT_FOUND
.value
)
50 except Exception as e
: # TODO refine
51 raise DbException(str(e
))
53 def del_list(self
, table
, filter={}):
56 for i
, _
in self
._find
(table
, self
._format
_filter
(filter)):
58 deleted
= len(id_list
)
61 return {"deleted": deleted
}
64 except Exception as e
: # TODO refine
65 raise DbException(str(e
))
67 def del_one(self
, table
, filter={}, fail_on_empty
=True):
69 for i
, _
in self
._find
(table
, self
._format
_filter
(filter)):
73 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus
.NOT_FOUND
.value
)
77 except Exception as e
: # TODO refine
78 raise DbException(str(e
))
80 def replace(self
, table
, filter, indata
, fail_on_empty
=True):
82 for i
, _
in self
._find
(table
, self
._format
_filter
(filter)):
86 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus
.NOT_FOUND
.value
)
88 self
.db
[table
][i
] = deepcopy(indata
)
90 except Exception as e
: # TODO refine
91 raise DbException(str(e
))
93 def create(self
, table
, indata
):
95 id = indata
.get("_id")
99 if table
not in self
.db
:
101 self
.db
[table
].append(deepcopy(indata
))
103 except Exception as e
: # TODO refine
104 raise DbException(str(e
))
107 if __name__
== '__main__':
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}))