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.__dbbackend.set(ns, data_map)
 147
 148     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, old_data=bytes, new_data=bytes)
 149     def set_if(self, ns: str, key: str, old_data: bytes, new_data: bytes) -> bool:
 150         return self.__dbbackend.set_if(ns, key, old_data, new_data)
 151
 152     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 153     def set_if_not_exists(self, ns: str, key: str, data: bytes) -> bool:
 154         return self.__dbbackend.set_if_not_exists(ns, key, data)
 155
 156     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 157     def get(self, ns: str, keys: Union[str, Set[str]]) -> Dict[str, bytes]:
 158         disordered = self.__dbbackend.get(ns, list(keys))
 159         return {k: disordered[k] for k in sorted(disordered)}
 160
 161     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 162     def find_keys(self, ns: str, key_pattern: str) -> List[str]:
 163         return self.__dbbackend.find_keys(ns, key_pattern)
 164
 165     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 166     def find_and_get(self, ns: str, key_pattern: str) -> Dict[str, bytes]:
 167         disordered = self.__dbbackend.find_and_get(ns, key_pattern)
 168         return {k: disordered[k] for k in sorted(disordered)}
 169
 170     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 171     def remove(self, ns: str, keys: Union[str, Set[str]]) -> None:
 172         self.__dbbackend.remove(ns, list(keys))
 173
 174     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 175     def remove_if(self, ns: str, key: str, data: bytes) -> bool:
 176         return self.__dbbackend.remove_if(ns, key, data)
 177
 178     @func_arg_checker(SdlTypeError, 1, ns=str)
 179     def remove_all(self, ns: str) -> None:
 180         keys = self.__dbbackend.find_keys(ns, '*')
 181         if keys:
 182             self.__dbbackend.remove(ns, keys)
 183
 184     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 185     def add_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 186         self.__dbbackend.add_member(ns, group, members)
 187
 188     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 189     def remove_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 190         self.__dbbackend.remove_member(ns, group, members)
 191
 192     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 193     def remove_group(self, ns: str, group: str) -> None:
 194         self.__dbbackend.remove_group(ns, group)
 195
 196     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 197     def get_members(self, ns: str, group: str) -> Set[bytes]:
 198         return self.__dbbackend.get_members(ns, group)
 199
 200     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, member=bytes)
 201     def is_member(self, ns: str, group: str, member: bytes) -> bool:
 202         return self.__dbbackend.is_member(ns, group, member)
 203
 204     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 205     def group_size(self, ns: str, group: str) -> int:
 206         return self.__dbbackend.group_size(ns, group)
 207
 208     @func_arg_checker(SdlTypeError, 1, ns=str, resource=str, expiration=(int, float))
 209     def get_lock_resource(self, ns: str, resource: str, expiration: Union[int, float]) -> SyncLock:
 210         return SyncLock(ns, resource, expiration, self)
 211
 212     def get_backend(self) -> DbBackendAbc:
 213         """Return backend instance."""
 214         return self.__dbbackend
 215
 216     def get_configuration(self) -> _Configuration:
 217         """Return configuration what was valid when the SDL instance was initiated."""
 218         return self.__configuration