ricsdl-package/ricsdl/syncstorage.py

   1 # Copyright (c) 2019 AT&T Intellectual Property.
   2 # Copyright (c) 2018-2019 Nokia.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 #
  17 # This source code is part of the near-RT RIC (RAN Intelligent Controller)
  18 # platform project (RICP).
  19 #
  20
  21 """The module provides implementation of the syncronous Shared Data Layer (SDL) interface."""
  22 import builtins
  23 from typing import (Dict, Set, List, Union)
  24 from ricsdl.configuration import _Configuration
  25 from ricsdl.syncstorage_abc import (SyncStorageAbc, SyncLockAbc)
  26 import ricsdl.backend
  27 from ricsdl.backend.dbbackend_abc import DbBackendAbc
  28 from ricsdl.exceptions import SdlTypeError
  29
  30
  31 def func_arg_checker(exception, start_arg_idx, **types):
  32     """Decorator to validate function arguments."""
  33     def _check(func):
  34         if not __debug__:
  35             return func
  36
  37         def _validate(*args, **kwds):
  38             for idx, arg in enumerate(args[start_arg_idx:], start_arg_idx):
  39                 if func.__code__.co_varnames[idx] in types and \
  40                         not isinstance(arg, types[func.__code__.co_varnames[idx]]):
  41                     raise exception(r"Wrong argument type: '{}'={}. Must be: {}".
  42                                     format(func.__code__.co_varnames[idx], type(arg),
  43                                            types[func.__code__.co_varnames[idx]]))
  44             for kwdname, kwdval in kwds.items():
  45                 if kwdname in types and not isinstance(kwdval, types[kwdname]):
  46                     raise exception(r"Wrong argument type: '{}'={}. Must be: {}".
  47                                     format(kwdname, type(kwdval), types[kwdname]))
  48             return func(*args, **kwds)
  49         _validate.__name__ = func.__name__
  50         return _validate
  51     return _check
  52
  53
  54 class SyncLock(SyncLockAbc):
  55     """
  56     This class implements Shared Data Layer (SDL) abstract 'SyncLockAbc' class.
  57
  58     A lock instance is created per namespace and it is identified by its `name` within a namespace.
  59
  60     Args:
  61         ns (str): Namespace under which this lock is targeted.
  62         name (str): Lock name, identifies the lock key in SDL storage.
  63         expiration (int, float): Lock expiration time after which the lock is removed if it hasn't
  64                                  been released earlier by a 'release' method.
  65         storage (SyncStorage): Database backend object containing connection to a database.
  66     """
  67     @func_arg_checker(SdlTypeError, 1, ns=str, name=str, expiration=(int, float))
  68     def __init__(self, ns: str, name: str, expiration: Union[int, float],
  69                  storage: 'SyncStorage') -> None:
  70
  71         super().__init__(ns, name, expiration)
  72         self.__configuration = storage.get_configuration()
  73         self.__dbbackendlock = ricsdl.backend.get_backend_lock_instance(self.__configuration,
  74                                                                         ns, name, expiration,
  75                                                                         storage.get_backend())
  76
  77     def __str__(self):
  78         return str(
  79             {
  80                 "namespace": self._ns,
  81                 "name": self._name,
  82                 "expiration": self._expiration,
  83                 "backend lock": str(self.__dbbackendlock)
  84             }
  85         )
  86
  87     @func_arg_checker(SdlTypeError, 1, retry_interval=(int, float),
  88                       retry_timeout=(int, float))
  89     def acquire(self, retry_interval: Union[int, float] = 0.1,
  90                 retry_timeout: Union[int, float] = 10) -> bool:
  91         return self.__dbbackendlock.acquire(retry_interval, retry_timeout)
  92
  93     def release(self) -> None:
  94         self.__dbbackendlock.release()
  95
  96     def refresh(self) -> None:
  97         self.__dbbackendlock.refresh()
  98
  99     def get_validity_time(self) -> Union[int, float]:
 100         return self.__dbbackendlock.get_validity_time()
 101
 102
 103 class SyncStorage(SyncStorageAbc):
 104     """
 105     This class implements Shared Data Layer (SDL) abstract 'SyncStorageAbc' class.
 106
 107     This class provides synchronous access to all the namespaces in SDL storage.
 108     Data can be written, read and removed based on keys known to clients. Keys are unique within
 109     a namespace, namespace identifier is passed as a parameter to all the operations.
 110
 111     Args:
 112         fake_db_backend (str): Optional parameter. Parameter enables fake DB backend usage for an
 113                                SDL instance. Fake DB backend is ONLY allowed to use for testing
 114                                purposes at development phase of SDL clients when more advanced
 115                                database services are not necessarily needed. Currently value 'dict'
 116                                is only allowed value for the parameter, which enables dictionary
 117                                type of fake DB backend.
 118     """
 119     def __init__(self, fake_db_backend=None) -> None:
 120         super().__init__()
 121         self.__configuration = _Configuration(fake_db_backend)
 122         self.__dbbackend = ricsdl.backend.get_backend_instance(self.__configuration)
 123
 124     def __del__(self):
 125         self.close()
 126
 127     def __str__(self):
 128         return str(
 129             {
 130                 "configuration": str(self.__configuration),
 131                 "backend": str(self.__dbbackend)
 132             }
 133         )
 134
 135     def close(self):
 136         self.__dbbackend.close()
 137
 138     @func_arg_checker(SdlTypeError, 1, ns=str, data_map=dict)
 139     def set(self, ns: str, data_map: Dict[str, bytes]) -> None:
 140         self.__dbbackend.set(ns, data_map)
 141
 142     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, old_data=bytes, new_data=bytes)
 143     def set_if(self, ns: str, key: str, old_data: bytes, new_data: bytes) -> bool:
 144         return self.__dbbackend.set_if(ns, key, old_data, new_data)
 145
 146     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 147     def set_if_not_exists(self, ns: str, key: str, data: bytes) -> bool:
 148         return self.__dbbackend.set_if_not_exists(ns, key, data)
 149
 150     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 151     def get(self, ns: str, keys: Union[str, Set[str]]) -> Dict[str, bytes]:
 152         disordered = self.__dbbackend.get(ns, list(keys))
 153         return {k: disordered[k] for k in sorted(disordered)}
 154
 155     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 156     def find_keys(self, ns: str, key_pattern: str) -> List[str]:
 157         return self.__dbbackend.find_keys(ns, key_pattern)
 158
 159     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 160     def find_and_get(self, ns: str, key_pattern: str) -> Dict[str, bytes]:
 161         disordered = self.__dbbackend.find_and_get(ns, key_pattern)
 162         return {k: disordered[k] for k in sorted(disordered)}
 163
 164     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 165     def remove(self, ns: str, keys: Union[str, Set[str]]) -> None:
 166         self.__dbbackend.remove(ns, list(keys))
 167
 168     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 169     def remove_if(self, ns: str, key: str, data: bytes) -> bool:
 170         return self.__dbbackend.remove_if(ns, key, data)
 171
 172     @func_arg_checker(SdlTypeError, 1, ns=str)
 173     def remove_all(self, ns: str) -> None:
 174         keys = self.__dbbackend.find_keys(ns, '*')
 175         if keys:
 176             self.__dbbackend.remove(ns, keys)
 177
 178     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 179     def add_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 180         self.__dbbackend.add_member(ns, group, members)
 181
 182     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 183     def remove_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 184         self.__dbbackend.remove_member(ns, group, members)
 185
 186     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 187     def remove_group(self, ns: str, group: str) -> None:
 188         self.__dbbackend.remove_group(ns, group)
 189
 190     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 191     def get_members(self, ns: str, group: str) -> Set[bytes]:
 192         return self.__dbbackend.get_members(ns, group)
 193
 194     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, member=bytes)
 195     def is_member(self, ns: str, group: str, member: bytes) -> bool:
 196         return self.__dbbackend.is_member(ns, group, member)
 197
 198     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 199     def group_size(self, ns: str, group: str) -> int:
 200         return self.__dbbackend.group_size(ns, group)
 201
 202     @func_arg_checker(SdlTypeError, 1, ns=str, resource=str, expiration=(int, float))
 203     def get_lock_resource(self, ns: str, resource: str, expiration: Union[int, float]) -> SyncLock:
 204         return SyncLock(ns, resource, expiration, self)
 205
 206     def get_backend(self) -> DbBackendAbc:
 207         """Return backend instance."""
 208         return self.__dbbackend
 209
 210     def get_configuration(self) -> _Configuration:
 211         """Return configuration what was valid when the SDL instance was initiated."""
 212         return self.__configuration