osm_common/dbmemory.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Telefonica S.A.
   4 #
   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
   8 #
   9 #    http://www.apache.org/licenses/LICENSE-2.0
  10 #
  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
  14 # implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17
  18 from copy import deepcopy
  19 from http import HTTPStatus
  20 import logging
  21 from uuid import uuid4
  22
  23 from osm_common.dbbase import DbBase, DbException
  24 from osm_common.dbmongo import deep_update
  25
  26
  27 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  28
  29
  30 class DbMemory(DbBase):
  31     def __init__(self, logger_name="db", lock=False):
  32         super().__init__(logger_name=logger_name, lock=lock)
  33         self.db = {}
  34
  35     def db_connect(self, config):
  36         """
  37         Connect to database
  38         :param config: Configuration of database
  39         :return: None or raises DbException on error
  40         """
  41         if "logger_name" in config:
  42             self.logger = logging.getLogger(config["logger_name"])
  43         master_key = config.get("commonkey") or config.get("masterpassword")
  44         if master_key:
  45             self.set_secret_key(master_key)
  46
  47     @staticmethod
  48     def _format_filter(q_filter):
  49         db_filter = {}
  50         # split keys with ANYINDEX in this way:
  51         # {"A.B.ANYINDEX.C.D.ANYINDEX.E": v }  -> {"A.B.ANYINDEX": {"C.D.ANYINDEX": {"E": v}}}
  52         if q_filter:
  53             for k, v in q_filter.items():
  54                 db_v = v
  55                 kleft, _, kright = k.rpartition(".ANYINDEX.")
  56                 while kleft:
  57                     k = kleft + ".ANYINDEX"
  58                     db_v = {kright: db_v}
  59                     kleft, _, kright = k.rpartition(".ANYINDEX.")
  60                 deep_update(db_filter, {k: db_v})
  61
  62         return db_filter
  63
  64     def _find(self, table, q_filter):
  65         def recursive_find(key_list, key_next_index, content, oper, target):
  66             if key_next_index == len(key_list) or content is None:
  67                 try:
  68                     if oper in ("eq", "cont"):
  69                         if isinstance(target, list):
  70                             if isinstance(content, list):
  71                                 return any(
  72                                     content_item in target for content_item in content
  73                                 )
  74                             return content in target
  75                         elif isinstance(content, list):
  76                             return target in content
  77                         else:
  78                             return content == target
  79                     elif oper in ("neq", "ne", "ncont"):
  80                         if isinstance(target, list):
  81                             if isinstance(content, list):
  82                                 return all(
  83                                     content_item not in target
  84                                     for content_item in content
  85                                 )
  86                             return content not in target
  87                         elif isinstance(content, list):
  88                             return target not in content
  89                         else:
  90                             return content != target
  91                     if oper == "gt":
  92                         return content > target
  93                     elif oper == "gte":
  94                         return content >= target
  95                     elif oper == "lt":
  96                         return content < target
  97                     elif oper == "lte":
  98                         return content <= target
  99                     else:
 100                         raise DbException(
 101                             "Unknown filter operator '{}' in key '{}'".format(
 102                                 oper, ".".join(key_list)
 103                             ),
 104                             http_code=HTTPStatus.BAD_REQUEST,
 105                         )
 106                 except TypeError:
 107                     return False
 108
 109             elif isinstance(content, dict):
 110                 return recursive_find(
 111                     key_list,
 112                     key_next_index + 1,
 113                     content.get(key_list[key_next_index]),
 114                     oper,
 115                     target,
 116                 )
 117             elif isinstance(content, list):
 118                 look_for_match = True  # when there is a match return immediately
 119                 if (target is None) != (
 120                     oper in ("neq", "ne", "ncont")
 121                 ):  # one True and other False (Xor)
 122                     look_for_match = (
 123                         False  # when there is not a match return immediately
 124                     )
 125
 126                 for content_item in content:
 127                     if key_list[key_next_index] == "ANYINDEX" and isinstance(v, dict):
 128                         matches = True
 129                         if target:
 130                             for k2, v2 in target.items():
 131                                 k_new_list = k2.split(".")
 132                                 new_operator = "eq"
 133                                 if k_new_list[-1] in (
 134                                     "eq",
 135                                     "ne",
 136                                     "gt",
 137                                     "gte",
 138                                     "lt",
 139                                     "lte",
 140                                     "cont",
 141                                     "ncont",
 142                                     "neq",
 143                                 ):
 144                                     new_operator = k_new_list.pop()
 145                                 if not recursive_find(
 146                                     k_new_list, 0, content_item, new_operator, v2
 147                                 ):
 148                                     matches = False
 149                                     break
 150
 151                     else:
 152                         matches = recursive_find(
 153                             key_list, key_next_index, content_item, oper, target
 154                         )
 155                     if matches == look_for_match:
 156                         return matches
 157                 if key_list[key_next_index].isdecimal() and int(
 158                     key_list[key_next_index]
 159                 ) < len(content):
 160                     matches = recursive_find(
 161                         key_list,
 162                         key_next_index + 1,
 163                         content[int(key_list[key_next_index])],
 164                         oper,
 165                         target,
 166                     )
 167                     if matches == look_for_match:
 168                         return matches
 169                 return not look_for_match
 170             else:  # content is not dict, nor list neither None, so not found
 171                 if oper in ("neq", "ne", "ncont"):
 172                     return target is not None
 173                 else:
 174                     return target is None
 175
 176         for i, row in enumerate(self.db.get(table, ())):
 177             q_filter = q_filter or {}
 178             for k, v in q_filter.items():
 179                 k_list = k.split(".")
 180                 operator = "eq"
 181                 if k_list[-1] in (
 182                     "eq",
 183                     "ne",
 184                     "gt",
 185                     "gte",
 186                     "lt",
 187                     "lte",
 188                     "cont",
 189                     "ncont",
 190                     "neq",
 191                 ):
 192                     operator = k_list.pop()
 193                 matches = recursive_find(k_list, 0, row, operator, v)
 194                 if not matches:
 195                     break
 196             else:
 197                 # match
 198                 yield i, row
 199
 200     def get_list(self, table, q_filter=None):
 201         """
 202         Obtain a list of entries matching q_filter
 203         :param table: collection or table
 204         :param q_filter: Filter
 205         :return: a list (can be empty) with the found entries. Raises DbException on error
 206         """
 207         try:
 208             result = []
 209             with self.lock:
 210                 for _, row in self._find(table, self._format_filter(q_filter)):
 211                     result.append(deepcopy(row))
 212             return result
 213         except DbException:
 214             raise
 215         except Exception as e:  # TODO refine
 216             raise DbException(str(e))
 217
 218     def count(self, table, q_filter=None):
 219         """
 220         Count the number of entries matching q_filter
 221         :param table: collection or table
 222         :param q_filter: Filter
 223         :return: number of entries found (can be zero)
 224         :raise: DbException on error
 225         """
 226         try:
 227             with self.lock:
 228                 return sum(1 for x in self._find(table, self._format_filter(q_filter)))
 229         except DbException:
 230             raise
 231         except Exception as e:  # TODO refine
 232             raise DbException(str(e))
 233
 234     def get_one(self, table, q_filter=None, fail_on_empty=True, fail_on_more=True):
 235         """
 236         Obtain one entry matching q_filter
 237         :param table: collection or table
 238         :param q_filter: Filter
 239         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 240         it raises a DbException
 241         :param fail_on_more: If more than one matches filter it returns one of then unless this flag is set tu True, so
 242         that it raises a DbException
 243         :return: The requested element, or None
 244         """
 245         try:
 246             result = None
 247             with self.lock:
 248                 for _, row in self._find(table, self._format_filter(q_filter)):
 249                     if not fail_on_more:
 250                         return deepcopy(row)
 251                     if result:
 252                         raise DbException(
 253                             "Found more than one entry with filter='{}'".format(
 254                                 q_filter
 255                             ),
 256                             HTTPStatus.CONFLICT.value,
 257                         )
 258                     result = row
 259             if not result and fail_on_empty:
 260                 raise DbException(
 261                     "Not found entry with filter='{}'".format(q_filter),
 262                     HTTPStatus.NOT_FOUND,
 263                 )
 264             return deepcopy(result)
 265         except Exception as e:  # TODO refine
 266             raise DbException(str(e))
 267
 268     def del_list(self, table, q_filter=None):
 269         """
 270         Deletes all entries that match q_filter
 271         :param table: collection or table
 272         :param q_filter: Filter
 273         :return: Dict with the number of entries deleted
 274         """
 275         try:
 276             id_list = []
 277             with self.lock:
 278                 for i, _ in self._find(table, self._format_filter(q_filter)):
 279                     id_list.append(i)
 280             deleted = len(id_list)
 281             for i in reversed(id_list):
 282                 del self.db[table][i]
 283             return {"deleted": deleted}
 284         except DbException:
 285             raise
 286         except Exception as e:  # TODO refine
 287             raise DbException(str(e))
 288
 289     def del_one(self, table, q_filter=None, fail_on_empty=True):
 290         """
 291         Deletes one entry that matches q_filter
 292         :param table: collection or table
 293         :param q_filter: Filter
 294         :param fail_on_empty: If nothing matches filter it returns '0' deleted unless this flag is set tu True, in
 295         which case it raises a DbException
 296         :return: Dict with the number of entries deleted
 297         """
 298         try:
 299             with self.lock:
 300                 for i, _ in self._find(table, self._format_filter(q_filter)):
 301                     break
 302                 else:
 303                     if fail_on_empty:
 304                         raise DbException(
 305                             "Not found entry with filter='{}'".format(q_filter),
 306                             HTTPStatus.NOT_FOUND,
 307                         )
 308                     return None
 309                 del self.db[table][i]
 310             return {"deleted": 1}
 311         except Exception as e:  # TODO refine
 312             raise DbException(str(e))
 313
 314     def _update(
 315         self,
 316         db_item,
 317         update_dict,
 318         unset=None,
 319         pull=None,
 320         push=None,
 321         push_list=None,
 322         pull_list=None,
 323     ):
 324         """
 325         Modifies an entry at database
 326         :param db_item: entry of the table to update
 327         :param update_dict: Plain dictionary with the content to be updated. It is a dot separated keys and a value
 328         :param unset: Plain dictionary with the content to be removed if exist. It is a dot separated keys, value is
 329                       ignored. If not exist, it is ignored
 330         :param pull: Plain dictionary with the content to be removed from an array. It is a dot separated keys and value
 331                      if exist in the array is removed. If not exist, it is ignored
 332         :param pull_list: Same as pull but values are arrays where each item is removed from the array
 333         :param push: Plain dictionary with the content to be appended to an array. It is a dot separated keys and value
 334                      is appended to the end of the array
 335         :param push_list: Same as push but values are arrays where each item is and appended instead of appending the
 336                           whole array
 337         :return: True if database has been changed, False if not; Exception on error
 338         """
 339
 340         def _iterate_keys(k, db_nested, populate=True):
 341             k_list = k.split(".")
 342             k_item_prev = k_list[0]
 343             populated = False
 344             if k_item_prev not in db_nested and populate:
 345                 populated = True
 346                 db_nested[k_item_prev] = None
 347             for k_item in k_list[1:]:
 348                 if isinstance(db_nested[k_item_prev], dict):
 349                     if k_item not in db_nested[k_item_prev]:
 350                         if not populate:
 351                             raise DbException(
 352                                 "Cannot set '{}', not existing '{}'".format(k, k_item)
 353                             )
 354                         populated = True
 355                         db_nested[k_item_prev][k_item] = None
 356                 elif isinstance(db_nested[k_item_prev], list) and k_item.isdigit():
 357                     # extend list with Nones if index greater than list
 358                     k_item = int(k_item)
 359                     if k_item >= len(db_nested[k_item_prev]):
 360                         if not populate:
 361                             raise DbException(
 362                                 "Cannot set '{}', index too large '{}'".format(
 363                                     k, k_item
 364                                 )
 365                             )
 366                         populated = True
 367                         db_nested[k_item_prev] += [None] * (
 368                             k_item - len(db_nested[k_item_prev]) + 1
 369                         )
 370                 elif db_nested[k_item_prev] is None:
 371                     if not populate:
 372                         raise DbException(
 373                             "Cannot set '{}', not existing '{}'".format(k, k_item)
 374                         )
 375                     populated = True
 376                     db_nested[k_item_prev] = {k_item: None}
 377                 else:  # number, string, boolean, ... or list but with not integer key
 378                     raise DbException(
 379                         "Cannot set '{}' on existing '{}={}'".format(
 380                             k, k_item_prev, db_nested[k_item_prev]
 381                         )
 382                     )
 383                 db_nested = db_nested[k_item_prev]
 384                 k_item_prev = k_item
 385             return db_nested, k_item_prev, populated
 386
 387         updated = False
 388         try:
 389             if update_dict:
 390                 for dot_k, v in update_dict.items():
 391                     dict_to_update, key_to_update, _ = _iterate_keys(dot_k, db_item)
 392                     dict_to_update[key_to_update] = v
 393                     updated = True
 394             if unset:
 395                 for dot_k in unset:
 396                     try:
 397                         dict_to_update, key_to_update, _ = _iterate_keys(
 398                             dot_k, db_item, populate=False
 399                         )
 400                         del dict_to_update[key_to_update]
 401                         updated = True
 402                     except Exception as unset_error:
 403                         self.logger.error(f"{unset_error} occured while updating DB.")
 404             if pull:
 405                 for dot_k, v in pull.items():
 406                     try:
 407                         dict_to_update, key_to_update, _ = _iterate_keys(
 408                             dot_k, db_item, populate=False
 409                         )
 410                     except Exception as pull_error:
 411                         self.logger.error(f"{pull_error} occured while updating DB.")
 412                         continue
 413
 414                     if key_to_update not in dict_to_update:
 415                         continue
 416                     if not isinstance(dict_to_update[key_to_update], list):
 417                         raise DbException(
 418                             "Cannot pull '{}'. Target is not a list".format(dot_k)
 419                         )
 420                     while v in dict_to_update[key_to_update]:
 421                         dict_to_update[key_to_update].remove(v)
 422                         updated = True
 423             if pull_list:
 424                 for dot_k, v in pull_list.items():
 425                     if not isinstance(v, list):
 426                         raise DbException(
 427                             "Invalid content at pull_list, '{}' must be an array".format(
 428                                 dot_k
 429                             ),
 430                             http_code=HTTPStatus.BAD_REQUEST,
 431                         )
 432                     try:
 433                         dict_to_update, key_to_update, _ = _iterate_keys(
 434                             dot_k, db_item, populate=False
 435                         )
 436                     except Exception as iterate_error:
 437                         self.logger.error(
 438                             f"{iterate_error} occured while iterating keys in db update."
 439                         )
 440                         continue
 441
 442                     if key_to_update not in dict_to_update:
 443                         continue
 444                     if not isinstance(dict_to_update[key_to_update], list):
 445                         raise DbException(
 446                             "Cannot pull_list '{}'. Target is not a list".format(dot_k)
 447                         )
 448                     for single_v in v:
 449                         while single_v in dict_to_update[key_to_update]:
 450                             dict_to_update[key_to_update].remove(single_v)
 451                             updated = True
 452             if push:
 453                 for dot_k, v in push.items():
 454                     dict_to_update, key_to_update, populated = _iterate_keys(
 455                         dot_k, db_item
 456                     )
 457                     if (
 458                         isinstance(dict_to_update, dict)
 459                         and key_to_update not in dict_to_update
 460                     ):
 461                         dict_to_update[key_to_update] = [v]
 462                         updated = True
 463                     elif populated and dict_to_update[key_to_update] is None:
 464                         dict_to_update[key_to_update] = [v]
 465                         updated = True
 466                     elif not isinstance(dict_to_update[key_to_update], list):
 467                         raise DbException(
 468                             "Cannot push '{}'. Target is not a list".format(dot_k)
 469                         )
 470                     else:
 471                         dict_to_update[key_to_update].append(v)
 472                         updated = True
 473             if push_list:
 474                 for dot_k, v in push_list.items():
 475                     if not isinstance(v, list):
 476                         raise DbException(
 477                             "Invalid content at push_list, '{}' must be an array".format(
 478                                 dot_k
 479                             ),
 480                             http_code=HTTPStatus.BAD_REQUEST,
 481                         )
 482                     dict_to_update, key_to_update, populated = _iterate_keys(
 483                         dot_k, db_item
 484                     )
 485                     if (
 486                         isinstance(dict_to_update, dict)
 487                         and key_to_update not in dict_to_update
 488                     ):
 489                         dict_to_update[key_to_update] = v.copy()
 490                         updated = True
 491                     elif populated and dict_to_update[key_to_update] is None:
 492                         dict_to_update[key_to_update] = v.copy()
 493                         updated = True
 494                     elif not isinstance(dict_to_update[key_to_update], list):
 495                         raise DbException(
 496                             "Cannot push '{}'. Target is not a list".format(dot_k),
 497                             http_code=HTTPStatus.CONFLICT,
 498                         )
 499                     else:
 500                         dict_to_update[key_to_update] += v
 501                         updated = True
 502
 503             return updated
 504         except DbException:
 505             raise
 506         except Exception as e:  # TODO refine
 507             raise DbException(str(e))
 508
 509     def set_one(
 510         self,
 511         table,
 512         q_filter,
 513         update_dict,
 514         fail_on_empty=True,
 515         unset=None,
 516         pull=None,
 517         push=None,
 518         push_list=None,
 519         pull_list=None,
 520     ):
 521         """
 522         Modifies an entry at database
 523         :param table: collection or table
 524         :param q_filter: Filter
 525         :param update_dict: Plain dictionary with the content to be updated. It is a dot separated keys and a value
 526         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 527         it raises a DbException
 528         :param unset: Plain dictionary with the content to be removed if exist. It is a dot separated keys, value is
 529                       ignored. If not exist, it is ignored
 530         :param pull: Plain dictionary with the content to be removed from an array. It is a dot separated keys and value
 531                      if exist in the array is removed. If not exist, it is ignored
 532         :param pull_list: Same as pull but values are arrays where each item is removed from the array
 533         :param push: Plain dictionary with the content to be appended to an array. It is a dot separated keys and value
 534                      is appended to the end of the array
 535         :param push_list: Same as push but values are arrays where each item is and appended instead of appending the
 536                           whole array
 537         :return: Dict with the number of entries modified. None if no matching is found.
 538         """
 539         with self.lock:
 540             for i, db_item in self._find(table, self._format_filter(q_filter)):
 541                 updated = self._update(
 542                     db_item,
 543                     update_dict,
 544                     unset=unset,
 545                     pull=pull,
 546                     push=push,
 547                     push_list=push_list,
 548                     pull_list=pull_list,
 549                 )
 550                 return {"updated": 1 if updated else 0}
 551             else:
 552                 if fail_on_empty:
 553                     raise DbException(
 554                         "Not found entry with _id='{}'".format(q_filter),
 555                         HTTPStatus.NOT_FOUND,
 556                     )
 557                 return None
 558
 559     def set_list(
 560         self,
 561         table,
 562         q_filter,
 563         update_dict,
 564         unset=None,
 565         pull=None,
 566         push=None,
 567         push_list=None,
 568         pull_list=None,
 569     ):
 570         """Modifies al matching entries at database. Same as push. Do not fail if nothing matches"""
 571         with self.lock:
 572             updated = 0
 573             found = 0
 574             for _, db_item in self._find(table, self._format_filter(q_filter)):
 575                 found += 1
 576                 if self._update(
 577                     db_item,
 578                     update_dict,
 579                     unset=unset,
 580                     pull=pull,
 581                     push=push,
 582                     push_list=push_list,
 583                     pull_list=pull_list,
 584                 ):
 585                     updated += 1
 586             # if not found and fail_on_empty:
 587             #     raise DbException("Not found entry with '{}'".format(q_filter), HTTPStatus.NOT_FOUND)
 588             return {"updated": updated} if found else None
 589
 590     def replace(self, table, _id, indata, fail_on_empty=True):
 591         """
 592         Replace the content of an entry
 593         :param table: collection or table
 594         :param _id: internal database id
 595         :param indata: content to replace
 596         :param fail_on_empty: If nothing matches filter it returns None unless this flag is set tu True, in which case
 597         it raises a DbException
 598         :return: Dict with the number of entries replaced
 599         """
 600         try:
 601             with self.lock:
 602                 for i, _ in self._find(table, self._format_filter({"_id": _id})):
 603                     break
 604                 else:
 605                     if fail_on_empty:
 606                         raise DbException(
 607                             "Not found entry with _id='{}'".format(_id),
 608                             HTTPStatus.NOT_FOUND,
 609                         )
 610                     return None
 611                 self.db[table][i] = deepcopy(indata)
 612             return {"updated": 1}
 613         except DbException:
 614             raise
 615         except Exception as e:  # TODO refine
 616             raise DbException(str(e))
 617
 618     def create(self, table, indata):
 619         """
 620         Add a new entry at database
 621         :param table: collection or table
 622         :param indata: content to be added
 623         :return: database '_id' of the inserted element. Raises a DbException on error
 624         """
 625         try:
 626             id = indata.get("_id")
 627             if not id:
 628                 id = str(uuid4())
 629                 indata["_id"] = id
 630             with self.lock:
 631                 if table not in self.db:
 632                     self.db[table] = []
 633                 self.db[table].append(deepcopy(indata))
 634             return id
 635         except Exception as e:  # TODO refine
 636             raise DbException(str(e))
 637
 638     def create_list(self, table, indata_list):
 639         """
 640         Add a new entry at database
 641         :param table: collection or table
 642         :param indata_list: list content to be added
 643         :return: list of inserted 'id's. Raises a DbException on error
 644         """
 645         try:
 646             _ids = []
 647             with self.lock:
 648                 for indata in indata_list:
 649                     _id = indata.get("_id")
 650                     if not _id:
 651                         _id = str(uuid4())
 652                         indata["_id"] = _id
 653                     with self.lock:
 654                         if table not in self.db:
 655                             self.db[table] = []
 656                         self.db[table].append(deepcopy(indata))
 657                     _ids.append(_id)
 658             return _ids
 659         except Exception as e:  # TODO refine
 660             raise DbException(str(e))
 661
 662
 663 if __name__ == "__main__":
 664     # some test code
 665     db = DbMemory()
 666     db.create("test", {"_id": 1, "data": 1})
 667     db.create("test", {"_id": 2, "data": 2})
 668     db.create("test", {"_id": 3, "data": 3})
 669     print("must be 3 items:", db.get_list("test"))
 670     print("must return item 2:", db.get_list("test", {"_id": 2}))
 671     db.del_one("test", {"_id": 2})
 672     print("must be emtpy:", db.get_list("test", {"_id": 2}))