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.__dbbackendlock = ricsdl.backend.get_backend_lock_instance(ns, name, expiration,
  73                                                                         storage.get_backend())
  74
  75     def __str__(self):
  76         return str(
  77             {
  78                 "namespace": self._ns,
  79                 "name": self._name,
  80                 "expiration": self._expiration,
  81                 "backend lock": str(self.__dbbackendlock)
  82             }
  83         )
  84
  85     @func_arg_checker(SdlTypeError, 1, retry_interval=(int, float),
  86                       retry_timeout=(int, float))
  87     def acquire(self, retry_interval: Union[int, float] = 0.1,
  88                 retry_timeout: Union[int, float] = 10) -> bool:
  89         return self.__dbbackendlock.acquire(retry_interval, retry_timeout)
  90
  91     def release(self) -> None:
  92         self.__dbbackendlock.release()
  93
  94     def refresh(self) -> None:
  95         self.__dbbackendlock.refresh()
  96
  97     def get_validity_time(self) -> Union[int, float]:
  98         return self.__dbbackendlock.get_validity_time()
  99
 100
 101 class SyncStorage(SyncStorageAbc):
 102     """
 103     This class implements Shared Data Layer (SDL) abstract 'SyncStorageAbc' class.
 104
 105     This class provides synchronous access to all the namespaces in SDL storage.
 106     Data can be written, read and removed based on keys known to clients. Keys are unique within
 107     a namespace, namespace identifier is passed as a parameter to all the operations.
 108
 109     Args:
 110         None
 111     """
 112     def __init__(self) -> None:
 113         super().__init__()
 114         self.__configuration = _Configuration()
 115         self.__dbbackend = ricsdl.backend.get_backend_instance(self.__configuration)
 116
 117     def __del__(self):
 118         self.close()
 119
 120     def __str__(self):
 121         return str(
 122             {
 123                 "configuration": str(self.__configuration),
 124                 "backend": str(self.__dbbackend)
 125             }
 126         )
 127
 128     def close(self):
 129         self.__dbbackend.close()
 130
 131     @func_arg_checker(SdlTypeError, 1, ns=str, data_map=dict)
 132     def set(self, ns: str, data_map: Dict[str, bytes]) -> None:
 133         self.__dbbackend.set(ns, data_map)
 134
 135     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, old_data=bytes, new_data=bytes)
 136     def set_if(self, ns: str, key: str, old_data: bytes, new_data: bytes) -> bool:
 137         return self.__dbbackend.set_if(ns, key, old_data, new_data)
 138
 139     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 140     def set_if_not_exists(self, ns: str, key: str, data: bytes) -> bool:
 141         return self.__dbbackend.set_if_not_exists(ns, key, data)
 142
 143     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 144     def get(self, ns: str, keys: Union[str, Set[str]]) -> Dict[str, bytes]:
 145         disordered = self.__dbbackend.get(ns, list(keys))
 146         return {k: disordered[k] for k in sorted(disordered)}
 147
 148     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 149     def find_keys(self, ns: str, key_pattern: str) -> List[str]:
 150         return self.__dbbackend.find_keys(ns, key_pattern)
 151
 152     @func_arg_checker(SdlTypeError, 1, ns=str, key_pattern=str)
 153     def find_and_get(self, ns: str, key_pattern: str) -> Dict[str, bytes]:
 154         disordered = self.__dbbackend.find_and_get(ns, key_pattern)
 155         return {k: disordered[k] for k in sorted(disordered)}
 156
 157     @func_arg_checker(SdlTypeError, 1, ns=str, keys=(str, builtins.set))
 158     def remove(self, ns: str, keys: Union[str, Set[str]]) -> None:
 159         self.__dbbackend.remove(ns, list(keys))
 160
 161     @func_arg_checker(SdlTypeError, 1, ns=str, key=str, data=bytes)
 162     def remove_if(self, ns: str, key: str, data: bytes) -> bool:
 163         return self.__dbbackend.remove_if(ns, key, data)
 164
 165     @func_arg_checker(SdlTypeError, 1, ns=str)
 166     def remove_all(self, ns: str) -> None:
 167         keys = self.__dbbackend.find_keys(ns, '*')
 168         if keys:
 169             self.__dbbackend.remove(ns, keys)
 170
 171     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 172     def add_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 173         self.__dbbackend.add_member(ns, group, members)
 174
 175     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, members=(bytes, builtins.set))
 176     def remove_member(self, ns: str, group: str, members: Union[bytes, Set[bytes]]) -> None:
 177         self.__dbbackend.remove_member(ns, group, members)
 178
 179     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 180     def remove_group(self, ns: str, group: str) -> None:
 181         self.__dbbackend.remove_group(ns, group)
 182
 183     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 184     def get_members(self, ns: str, group: str) -> Set[bytes]:
 185         return self.__dbbackend.get_members(ns, group)
 186
 187     @func_arg_checker(SdlTypeError, 1, ns=str, group=str, member=bytes)
 188     def is_member(self, ns: str, group: str, member: bytes) -> bool:
 189         return self.__dbbackend.is_member(ns, group, member)
 190
 191     @func_arg_checker(SdlTypeError, 1, ns=str, group=str)
 192     def group_size(self, ns: str, group: str) -> int:
 193         return self.__dbbackend.group_size(ns, group)
 194
 195     @func_arg_checker(SdlTypeError, 1, ns=str, resource=str, expiration=(int, float))
 196     def get_lock_resource(self, ns: str, resource: str, expiration: Union[int, float]) -> SyncLock:
 197         return SyncLock(ns, resource, expiration, self)
 198
 199     def get_backend(self) -> DbBackendAbc:
 200         """Return backend instance."""
 201         return self.__dbbackend