2 from dbbase
import DbException
, DbBase
3 from http
import HTTPStatus
5 from copy
import deepcopy
7 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
10 class DbMemory(DbBase
):
12 def __init__(self
, logger_name
='db'):
13 self
.logger
= logging
.getLogger(logger_name
)
16 def db_connect(self
, config
):
17 if "logger_name" in config
:
18 self
.logger
= logging
.getLogger(config
["logger_name"])
21 def _format_filter(filter):
24 def _find(self
, table
, filter):
25 for i
, row
in enumerate(self
.db
.get(table
, ())):
28 for k
, v
in filter.items():
29 if k
not in row
or v
!= row
[k
]:
34 def get_list(self
, table
, filter={}):
37 for _
, row
in self
._find
(table
, self
._format
_filter
(filter)):
38 l
.append(deepcopy(row
))
42 except Exception as e
: # TODO refine
43 raise DbException(str(e
))
45 def get_one(self
, table
, filter={}, fail_on_empty
=True, fail_on_more
=True):
48 for _
, row
in self
._find
(table
, self
._format
_filter
(filter)):
52 raise DbException("Found more than one entry with filter='{}'".format(filter),
53 HTTPStatus
.CONFLICT
.value
)
55 if not l
and fail_on_empty
:
56 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus
.NOT_FOUND
)
58 except Exception as e
: # TODO refine
59 raise DbException(str(e
))
61 def del_list(self
, table
, filter={}):
64 for i
, _
in self
._find
(table
, self
._format
_filter
(filter)):
66 deleted
= len(id_list
)
69 return {"deleted": deleted
}
72 except Exception as e
: # TODO refine
73 raise DbException(str(e
))
75 def del_one(self
, table
, filter={}, fail_on_empty
=True):
77 for i
, _
in self
._find
(table
, self
._format
_filter
(filter)):
81 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus
.NOT_FOUND
)
85 except Exception as e
: # TODO refine
86 raise DbException(str(e
))
88 def replace(self
, table
, filter, indata
, fail_on_empty
=True):
90 for i
, _
in self
._find
(table
, self
._format
_filter
(filter)):
94 raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus
.NOT_FOUND
)
96 self
.db
[table
][i
] = deepcopy(indata
)
98 except Exception as e
: # TODO refine
99 raise DbException(str(e
))
101 def create(self
, table
, indata
):
103 id = indata
.get("_id")
107 if table
not in self
.db
:
109 self
.db
[table
].append(deepcopy(indata
))
111 except Exception as e
: # TODO refine
112 raise DbException(str(e
))
115 if __name__
== '__main__':
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}))