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 (SdlException, 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 is_active(self):
 136         try:
 137             return self.__dbbackend.is_connected()
 138         except SdlException:
 139             return False
 140
 141     def close(self):
 142         self.__dbbackend.close()
 143
 144     @func_arg_checker(SdlTypeError, 1, ns=str, data_map=dict)
 145     def set(self, ns: str, data_map: Dict[str, bytes]) -> None:
 146         self._validate_key_value_dict(data_map)
 147         self.__dbbackend.set(ns, data_map)
 148
 149     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, old_data=bytes, new_data=bytes)
 150     def set_if(self, ns: str, key: str, old_data: bytes, new_data: bytes) -> bool:
 151         return self.__dbbackend.set_if(ns, key, old_data, new_data)
 152
 153     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 154     def set_if_not_exists(self, ns: str, key: str, data: bytes) -> bool:
 155         return self.__dbbackend.set_if_not_exists(ns, key, data)
 156
 157     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 158     def get(self, ns: str, keys: Union[str, Set[str]]) -> Dict[str, bytes]:
 159         disordered = self.__dbbackend.get(ns, list(keys))
 160         return {k: disordered[k] for k in sorted(disordered)}
 161
 162     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 163     def find_keys(self, ns: str, key_pattern: str) -> List[str]:
 164         return self.__dbbackend.find_keys(ns, key_pattern)
 165
 166     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 167     def find_and_get(self, ns: str, key_pattern: str) -> Dict[str, bytes]:
 168         disordered = self.__dbbackend.find_and_get(ns, key_pattern)
 169         return {k: disordered[k] for k in sorted(disordered)}
 170
 171     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 172     def remove(self, ns: str, keys: Union[str, Set[str]]) -> None:
 173         self.__dbbackend.remove(ns, list(keys))
 174
 175     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 176     def remove_if(self, ns: str, key: str, data: bytes) -> bool:
 177         return self.__dbbackend.remove_if(ns, key, data)
 178
 179     @func_arg_checker(SdlTypeError, 1, ns=str)
 180     def remove_all(self, ns: str) -> None:
 181         keys = self.__dbbackend.find_keys(ns, '*')
 182         if keys:
 183             self.__dbbackend.remove(ns, keys)
 184
 185     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 186     def add_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 187         self.__dbbackend.add_member(ns, group, members)
 188
 189     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 190     def remove_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 191         self.__dbbackend.remove_member(ns, group, members)
 192
 193     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 194     def remove_group(self, ns: str, group: str) -> None:
 195         self.__dbbackend.remove_group(ns, group)
 196
 197     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 198     def get_members(self, ns: str, group: str) -> Set[bytes]:
 199         return self.__dbbackend.get_members(ns, group)
 200
 201     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, member=bytes)
 202     def is_member(self, ns: str, group: str, member: bytes) -> bool:
 203         return self.__dbbackend.is_member(ns, group, member)
 204
 205     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 206     def group_size(self, ns: str, group: str) -> int:
 207         return self.__dbbackend.group_size(ns, group)
 208
 209     @func_arg_checker(SdlTypeError, 1, ns=str, resource=str, expiration=(int, float))
 210     def get_lock_resource(self, ns: str, resource: str, expiration: Union[int, float]) -> SyncLock:
 211         return SyncLock(ns, resource, expiration, self)
 212
 213     def get_backend(self) -> DbBackendAbc:
 214         """Return backend instance."""
 215         return self.__dbbackend
 216
 217     def get_configuration(self) -> _Configuration:
 218         """Return configuration what was valid when the SDL instance was initiated."""
 219         return self.__configuration
 220
 221     @classmethod
 222     def _validate_key_value_dict(cls, kv):
 223         for k, v in kv.items():
 224             if not isinstance(k, str):
 225                 raise SdlTypeError(r"Wrong dict key type: {}={}. Must be: str".format(k, type(k)))
 226             if not isinstance(v, bytes):
 227                 raise SdlTypeError(r"Wrong dict value type: {}={}. Must be: bytes".format(v, type(v)))