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 self
.master_password
= config
.get("masterpassword")
44 def _format_filter(q_filter
):
45 return q_filter
# TODO
47 def _find(self
, table
, q_filter
):
48 for i
, row
in enumerate(self
.db
.get(table
, ())):
51 for k
, v
in q_filter
.items():
52 if k
not in row
or v
!= row
[k
]:
57 def get_list(self
, table
, q_filter
=None):
59 Obtain a list of entries matching q_filter
60 :param table: collection or table
61 :param q_filter: Filter
62 :return: a list (can be empty) with the found entries. Raises DbException on error
67 for _
, row
in self
._find
(table
, self
._format
_filter
(q_filter
)):
68 result
.append(deepcopy(row
))
72 except Exception as e
: # TODO refine
73 raise DbException(str(e
))
75 def get_one(self
, table
, q_filter
=None, fail_on_empty
=True, fail_on_more
=True):
77 Obtain one entry matching q_filter
78 :param table: collection or table
79 :param q_filter: Filter
80 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
81 it raises a DbException
82 :param fail_on_more: If more than one matches filter it returns one of then unless this flag is set tu True, so
83 that it raises a DbException
84 :return: The requested element, or None
89 for _
, row
in self
._find
(table
, self
._format
_filter
(q_filter
)):
93 raise DbException("Found more than one entry with filter='{}'".format(q_filter
),
94 HTTPStatus
.CONFLICT
.value
)
96 if not result
and fail_on_empty
:
97 raise DbException("Not found entry with filter='{}'".format(q_filter
), HTTPStatus
.NOT_FOUND
)
98 return deepcopy(result
)
99 except Exception as e
: # TODO refine
100 raise DbException(str(e
))
102 def del_list(self
, table
, q_filter
=None):
104 Deletes all entries that match q_filter
105 :param table: collection or table
106 :param q_filter: Filter
107 :return: Dict with the number of entries deleted
112 for i
, _
in self
._find
(table
, self
._format
_filter
(q_filter
)):
114 deleted
= len(id_list
)
115 for i
in reversed(id_list
):
116 del self
.db
[table
][i
]
117 return {"deleted": deleted
}
120 except Exception as e
: # TODO refine
121 raise DbException(str(e
))
123 def del_one(self
, table
, q_filter
=None, fail_on_empty
=True):
125 Deletes one entry that matches q_filter
126 :param table: collection or table
127 :param q_filter: Filter
128 :param fail_on_empty: If nothing matches filter it returns '0' deleted unless this flag is set tu True, in
129 which case it raises a DbException
130 :return: Dict with the number of entries deleted
134 for i
, _
in self
._find
(table
, self
._format
_filter
(q_filter
)):
138 raise DbException("Not found entry with filter='{}'".format(q_filter
), HTTPStatus
.NOT_FOUND
)
140 del self
.db
[table
][i
]
141 return {"deleted": 1}
142 except Exception as e
: # TODO refine
143 raise DbException(str(e
))
145 def replace(self
, table
, _id
, indata
, fail_on_empty
=True):
147 Replace the content of an entry
148 :param table: collection or table
149 :param _id: internal database id
150 :param indata: content to replace
151 :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
152 it raises a DbException
153 :return: Dict with the number of entries replaced
157 for i
, _
in self
._find
(table
, self
._format
_filter
({"_id": _id
})):
161 raise DbException("Not found entry with _id='{}'".format(_id
), HTTPStatus
.NOT_FOUND
)
163 self
.db
[table
][i
] = deepcopy(indata
)
164 return {"updated": 1}
167 except Exception as e
: # TODO refine
168 raise DbException(str(e
))
170 def create(self
, table
, indata
):
172 Add a new entry at database
173 :param table: collection or table
174 :param indata: content to be added
175 :return: database id of the inserted element. Raises a DbException on error
178 id = indata
.get("_id")
183 if table
not in self
.db
:
185 self
.db
[table
].append(deepcopy(indata
))
187 except Exception as e
: # TODO refine
188 raise DbException(str(e
))
191 if __name__
== '__main__':
194 db
.create("test", {"_id": 1, "data": 1})
195 db
.create("test", {"_id": 2, "data": 2})
196 db
.create("test", {"_id": 3, "data": 3})
197 print("must be 3 items:", db
.get_list("test"))
198 print("must return item 2:", db
.get_list("test", {"_id": 2}))
199 db
.del_one("test", {"_id": 2})
200 print("must be emtpy:", db
.get_list("test", {"_id": 2}))