sms-client/python/onap-sms-client/onap-sms-client/__init__.py

   1 # Copyright 2018 Intel Corporation, Inc
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #       http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import requests
  16 import requests.exceptions
  17 import urlparse
  18
  19 name = "onap-sms-client"
  20
  21
  22 class InvalidRequestException(Exception):
  23     pass
  24
  25
  26 class InternalServerError(Exception):
  27     pass
  28
  29
  30 class UnexpectedError(Exception):
  31     pass
  32
  33
  34 class Client(object):
  35     """Python Client for Secret Management Service"""
  36
  37     def __init__(self, url='http://localhost:10443', timeout=30, cacert=None):
  38         """Creates a new SMS client instance
  39
  40         Args:
  41             url (str): Base URL with port pointing to the SMS service
  42             timeout (int): Number of seconds before aborting the connection
  43             cacert (str): Path to the cacert that will be used to verify
  44                 If this is None, verify will be False and the server cert
  45                 is not verified by the client.
  46         Returns:
  47             A Client object which can be used to interact with SMS service
  48         """
  49
  50         self.base_url = url
  51         self.timeout = timeout
  52         self.cacert = cacert
  53         self.session = requests.Session()
  54
  55         self._base_api_url = '/v1/sms'
  56
  57     def _urlJoin(self, *urls):
  58         """Joins given urls into a single url
  59
  60         Args:
  61             urls (str): url fragments to be combined.
  62
  63         Returns:
  64             str: Joined URL
  65         """
  66
  67         return '/'.join(urls)
  68
  69     def _raiseException(self, statuscode, errors=None):
  70         """ Handles Exception Raising based on statusCode
  71
  72         Args:
  73             statuscode (int): status code returned by the server
  74             errors (str): list of strings containing error messages
  75
  76         Returns:
  77             exception: An exception is raised based on error message
  78         """
  79
  80         if statuscode == 400:
  81             raise InvalidRequestException(errors)
  82         if statuscode == 500:
  83             raise InternalServerError(errors)
  84
  85         raise UnexpectedError(errors)
  86
  87     def _request(self, method, url, headers=None, **kwargs):
  88         """Handles request for all the client methods
  89
  90         Args:
  91             method (str): type of HTTP method (get, post or delete).
  92             url (str): api URL.
  93             headers (dict): custom headers if any.
  94             **kwargs: various args supported by requests library
  95
  96         Returns:
  97             requests.Response: An object containing status_code and returned
  98                                json data is returned here.
  99         """
 100
 101         if headers is None:
 102             headers = {
 103                 'content-type': "application/json",
 104                 'Accept': "application/json"
 105             }
 106
 107         # Verify the server or not based on the cacert argument
 108         if self.cacert is None:
 109             verify = False
 110         else:
 111             verify = self.cacert
 112
 113         url = urlparse.urljoin(self.base_url, url)
 114         response = self.session.request(method, url, headers=headers,
 115                                         allow_redirects=False, verify=verify,
 116                                         timeout=self.timeout, **kwargs)
 117
 118         errors = None
 119         if response.status_code >= 400 and response.status_code < 600:
 120             # Request Failed. Raise Exception.
 121             errors = response.text
 122             self._raiseException(response.status_code, errors)
 123
 124         return response
 125
 126     def getStatus(self):
 127         """Returns Status of SMS Service
 128
 129         Returns:
 130             bool: True or False depending on if SMS Service is ready.
 131         """
 132
 133         url = self._urlJoin(self._base_api_url, 'quorum', 'status')
 134
 135         response = self._request('get', url)
 136         return response.json()['sealstatus']
 137
 138     def createDomain(self, domainName):
 139         """Creates a Secret Domain
 140
 141         Args:
 142             domainName (str): Name of the secret domain to create
 143
 144         Returns:
 145             string: UUID of the created domain name
 146         """
 147
 148         domainName = domainName.strip()
 149         data = {"name": domainName}
 150         url = self._urlJoin(self._base_api_url, 'domain')
 151
 152         response = self._request('post', url, json=data)
 153         return response.json()['uuid']
 154
 155     def deleteDomain(self, domainName):
 156         """Deletes a Secret Domain
 157
 158         Args:
 159             domainName (str): Name of the secret domain to delete
 160
 161         Returns:
 162             bool: True. An exception will be raised if delete failed.
 163         """
 164
 165         domainName = domainName.strip()
 166         url = self._urlJoin(self._base_api_url, 'domain', domainName)
 167
 168         self._request('delete', url)
 169         return True
 170
 171     def getSecretNames(self, domainName):
 172         """Get all Secret Names in Domain
 173
 174         Args:
 175             domainName (str): Name of the secret domain
 176
 177         Returns:
 178             string[]: List of strings each corresponding to a
 179                       Secret's Name in this Domain.
 180         """
 181
 182         domainName = domainName.strip()
 183         url = self._urlJoin(self._base_api_url, 'domain', domainName,
 184                             'secret')
 185
 186         response = self._request('get', url)
 187         return response.json()['secretnames']
 188
 189     def storeSecret(self, domainName, secretName, values):
 190         """Store a Secret in given Domain
 191
 192         Args:
 193             domainName (str): Name of the secret domain
 194             secretName (str): Name for the Secret
 195             values (dict): A dict containing name-value pairs which
 196                            form the secret
 197
 198         Returns:
 199             bool: True. An exception will be raised if store failed.
 200         """
 201
 202         domainName = domainName.strip()
 203         secretName = secretName.strip()
 204         url = self._urlJoin(self._base_api_url, 'domain', domainName,
 205                             'secret')
 206
 207         if not isinstance(values, dict):
 208             raise TypeError('Input values is not a dictionary')
 209
 210         data = {"name": secretName, "values": values}
 211         self._request('post', url, json=data)
 212         return True
 213
 214     def getSecret(self, domainName, secretName):
 215         """Get a particular Secret from Domain.
 216
 217         Args:
 218             domainName (str): Name of the secret domain
 219             secretName (str): Name of the secret
 220
 221         Returns:
 222             dict: dictionary containing the name-value pairs
 223                   which form the secret
 224         """
 225
 226         domainName = domainName.strip()
 227         secretName = secretName.strip()
 228         url = self._urlJoin(self._base_api_url, 'domain', domainName,
 229                             'secret', secretName)
 230
 231         response = self._request('get', url)
 232         return response.json()['values']
 233
 234     def deleteSecret(self, domainName, secretName):
 235         """Delete a particular Secret from Domain.
 236
 237         Args:
 238             domainName (str): Name of the secret domain
 239             secretName (str): Name of the secret
 240
 241         Returns:
 242             bool: True. An exception will be raised if delete failed.
 243         """
 244
 245         domainName = domainName.strip()
 246         secretName = secretName.strip()
 247         url = self._urlJoin(self._base_api_url, 'domain', domainName,
 248                             'secret', secretName)
 249
 250         self._request('delete', url)
 251         return True