# See the License for the specific language governing permissions and
# limitations under the License.
-import logging
-from osm_common.dbbase import DbException, DbBase
-from osm_common.dbmongo import deep_update
+from copy import deepcopy
from http import HTTPStatus
+import logging
from uuid import uuid4
-from copy import deepcopy
+
+from osm_common.dbbase import DbBase, DbException
+from osm_common.dbmongo import deep_update
+
__author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
class DbMemory(DbBase):
def __init__(self, logger_name="db", lock=False):
- super().__init__(logger_name, lock)
+ super().__init__(logger_name=logger_name, lock=lock)
self.db = {}
def db_connect(self, config):
)
del dict_to_update[key_to_update]
updated = True
- except Exception:
- pass
+ except Exception as unset_error:
+ self.logger.error(f"{unset_error} occured while updating DB.")
if pull:
for dot_k, v in pull.items():
try:
dict_to_update, key_to_update, _ = _iterate_keys(
dot_k, db_item, populate=False
)
- except Exception:
+ except Exception as pull_error:
+ self.logger.error(f"{pull_error} occured while updating DB.")
continue
+
if key_to_update not in dict_to_update:
continue
if not isinstance(dict_to_update[key_to_update], list):
dict_to_update, key_to_update, _ = _iterate_keys(
dot_k, db_item, populate=False
)
- except Exception:
+ except Exception as iterate_error:
+ self.logger.error(
+ f"{iterate_error} occured while iterating keys in db update."
+ )
continue
+
if key_to_update not in dict_to_update:
continue
if not isinstance(dict_to_update[key_to_update], list):