1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Telefonica S.A.
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
19 from osm_common
.dbbase
import DbException
, DbBase
20 from http
import HTTPStatus
21 from uuid
import uuid4
22 from copy
import deepcopy
24 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
27 class DbMemory(DbBase
):
29 def __init__(self
, logger_name
='db', lock
=False):
30 super().__init
__(logger_name
, lock
)
33 def db_connect(self
, config
):
36 :param config: Configuration of database
37 :return: None or raises DbException on error
39 if "logger_name" in config
:
40 self
.logger
= logging
.getLogger(config
["logger_name"])
41 master_key
= config
.get("commonkey") or config
.get("masterpassword")
43 self
.set_secret_key(master_key
)
46 def _format_filter(q_filter
):
47 return q_filter
# TODO
49 def _find(self
, table
, q_filter
):
50 for i
, row
in enumerate(self
.db
.get(table
, ())):
53 for k
, v
in q_filter
.items():
54 if k
not in row
or v
!= row
[k
]:
59 def get_list(self
, table
, q_filter
=None):
61 Obtain a list of entries matching q_filter
62 :param table: collection or table
63 :param q_filter: Filter
64 :return: a list (can be empty) with the found entries. Raises DbException on error
69 for _
, row
in self
._find
(table
, self
._format
_filter
(q_filter
)):
70 result
.append(deepcopy(row
))
74 except Exception as e
: # TODO refine
75 raise DbException(str(e
))
77 def get_one(self
, table
, q_filter
=None, fail_on_empty
=True, fail_on_more
=True):
79 Obtain one entry matching q_filter
80 :param table: collection or table
81 :param q_filter: Filter
82 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
83 it raises a DbException
84 :param fail_on_more: If more than one matches filter it returns one of then unless this flag is set tu True, so
85 that it raises a DbException
86 :return: The requested element, or None
91 for _
, row
in self
._find
(table
, self
._format
_filter
(q_filter
)):
95 raise DbException("Found more than one entry with filter='{}'".format(q_filter
),
96 HTTPStatus
.CONFLICT
.value
)
98 if not result
and fail_on_empty
:
99 raise DbException("Not found entry with filter='{}'".format(q_filter
), HTTPStatus
.NOT_FOUND
)
100 return deepcopy(result
)
101 except Exception as e
: # TODO refine
102 raise DbException(str(e
))
104 def del_list(self
, table
, q_filter
=None):
106 Deletes all entries that match q_filter
107 :param table: collection or table
108 :param q_filter: Filter
109 :return: Dict with the number of entries deleted
114 for i
, _
in self
._find
(table
, self
._format
_filter
(q_filter
)):
116 deleted
= len(id_list
)
117 for i
in reversed(id_list
):
118 del self
.db
[table
][i
]
119 return {"deleted": deleted
}
122 except Exception as e
: # TODO refine
123 raise DbException(str(e
))
125 def del_one(self
, table
, q_filter
=None, fail_on_empty
=True):
127 Deletes one entry that matches q_filter
128 :param table: collection or table
129 :param q_filter: Filter
130 :param fail_on_empty: If nothing matches filter it returns '0' deleted unless this flag is set tu True, in
131 which case it raises a DbException
132 :return: Dict with the number of entries deleted
136 for i
, _
in self
._find
(table
, self
._format
_filter
(q_filter
)):
140 raise DbException("Not found entry with filter='{}'".format(q_filter
), HTTPStatus
.NOT_FOUND
)
142 del self
.db
[table
][i
]
143 return {"deleted": 1}
144 except Exception as e
: # TODO refine
145 raise DbException(str(e
))
147 def replace(self
, table
, _id
, indata
, fail_on_empty
=True):
149 Replace the content of an entry
150 :param table: collection or table
151 :param _id: internal database id
152 :param indata: content to replace
153 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
154 it raises a DbException
155 :return: Dict with the number of entries replaced
159 for i
, _
in self
._find
(table
, self
._format
_filter
({"_id": _id
})):
163 raise DbException("Not found entry with _id='{}'".format(_id
), HTTPStatus
.NOT_FOUND
)
165 self
.db
[table
][i
] = deepcopy(indata
)
166 return {"updated": 1}
169 except Exception as e
: # TODO refine
170 raise DbException(str(e
))
172 def create(self
, table
, indata
):
174 Add a new entry at database
175 :param table: collection or table
176 :param indata: content to be added
177 :return: database id of the inserted element. Raises a DbException on error
180 id = indata
.get("_id")
185 if table
not in self
.db
:
187 self
.db
[table
].append(deepcopy(indata
))
189 except Exception as e
: # TODO refine
190 raise DbException(str(e
))
193 if __name__
== '__main__':
196 db
.create("test", {"_id": 1, "data": 1})
197 db
.create("test", {"_id": 2, "data": 2})
198 db
.create("test", {"_id": 3, "data": 3})
199 print("must be 3 items:", db
.get_list("test"))
200 print("must return item 2:", db
.get_list("test", {"_id": 2}))
201 db
.del_one("test", {"_id": 2})
202 print("must be emtpy:", db
.get_list("test", {"_id": 2}))