osmclient/sol005/http.py

   1 # Copyright 2018 Telefonica
   2 #
   3 # All Rights Reserved.
   4 #
   5 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   6 #    not use this file except in compliance with the License. You may obtain
   7 #    a copy of the License at
   8 #
   9 #         http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #    Unless required by applicable law or agreed to in writing, software
  12 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  13 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  14 #    License for the specific language governing permissions and limitations
  15 #    under the License.
  16
  17 from io import BytesIO
  18 import pycurl
  19 import json
  20 import logging
  21 import copy
  22 from osmclient.common import http
  23 from osmclient.common.exceptions import OsmHttpException, NotFound
  24
  25
  26 class Http(http.Http):
  27     CONNECT_TIMEOUT = 15
  28
  29     def __init__(self, url, user='admin', password='admin', **kwargs):
  30         self._url = url
  31         self._user = user
  32         self._password = password
  33         self._http_header = None
  34         self._logger = logging.getLogger('osmclient')
  35         self._default_query_admin = None
  36         self._all_projects = None;
  37         self._public = None;
  38         if 'all_projects' in kwargs:
  39             self._all_projects=kwargs['all_projects']
  40         if 'public' in kwargs:
  41             self._public=kwargs['public']
  42         self._default_query_admin = self._complete_default_query_admin()
  43
  44     def _complete_default_query_admin(self):
  45         query_string_list=[]
  46         if self._all_projects:
  47             query_string_list.append("ADMIN")
  48         if self._public is not None:
  49             query_string_list.append("PUBLIC={}".format(self._public))
  50         return "&".join(query_string_list)
  51
  52     def _complete_endpoint(self, endpoint):
  53         if self._default_query_admin:
  54             if '?' in endpoint:
  55                 endpoint = '&'.join([endpoint, self._default_query_admin])
  56             else:
  57                 endpoint = '?'.join([endpoint, self._default_query_admin])
  58         return endpoint
  59
  60     def _get_curl_cmd(self, endpoint, skip_query_admin=False):
  61         self._logger.debug("")
  62         curl_cmd = pycurl.Curl()
  63         if self._logger.getEffectiveLevel() == logging.DEBUG:
  64             curl_cmd.setopt(pycurl.VERBOSE, True)
  65         if not skip_query_admin:
  66             endpoint = self._complete_endpoint(endpoint)
  67         curl_cmd.setopt(pycurl.CONNECTTIMEOUT, self.CONNECT_TIMEOUT)
  68         curl_cmd.setopt(pycurl.URL, self._url + endpoint)
  69         curl_cmd.setopt(pycurl.SSL_VERIFYPEER, 0)
  70         curl_cmd.setopt(pycurl.SSL_VERIFYHOST, 0)
  71         if self._http_header:
  72             curl_cmd.setopt(pycurl.HTTPHEADER, self._http_header)
  73         return curl_cmd
  74
  75     def delete_cmd(self, endpoint, skip_query_admin=False):
  76         self._logger.debug("")
  77         data = BytesIO()
  78         curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
  79         curl_cmd.setopt(pycurl.CUSTOMREQUEST, "DELETE")
  80         curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
  81         self._logger.info("Request METHOD: {} URL: {}".format("DELETE",self._url + endpoint))
  82         curl_cmd.perform()
  83         http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
  84         self._logger.info("Response HTTPCODE: {}".format(http_code))
  85         curl_cmd.close()
  86         self.check_http_response(http_code, data)
  87         # TODO 202 accepted should be returned somehow
  88         if data.getvalue():
  89             data_text = data.getvalue().decode()
  90             self._logger.verbose("Response DATA: {}".format(data_text))
  91             return http_code, data_text
  92         else:
  93             return http_code, None
  94
  95     def send_cmd(self, endpoint='', postfields_dict=None,
  96                  formfile=None, filename=None,
  97                  put_method=False, patch_method=False,
  98                  skip_query_admin=False):
  99         self._logger.debug("")
 100         data = BytesIO()
 101         curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
 102         if put_method:
 103             curl_cmd.setopt(pycurl.CUSTOMREQUEST, "PUT")
 104         elif patch_method:
 105             curl_cmd.setopt(pycurl.CUSTOMREQUEST, "PATCH")
 106         curl_cmd.setopt(pycurl.POST, 1)
 107         curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
 108
 109         if postfields_dict is not None:
 110             jsondata = json.dumps(postfields_dict)
 111             if 'password' in postfields_dict:
 112                 postfields_dict_copy = copy.deepcopy(postfields_dict)
 113                 postfields_dict_copy['password']='******'
 114                 jsondata_log = json.dumps(postfields_dict_copy)
 115             else:
 116                 jsondata_log = jsondata
 117             self._logger.verbose("Request POSTFIELDS: {}".format(jsondata_log))
 118             curl_cmd.setopt(pycurl.POSTFIELDS, jsondata)
 119         elif formfile is not None:
 120             curl_cmd.setopt(
 121                 pycurl.HTTPPOST,
 122                 [((formfile[0],
 123                            (pycurl.FORM_FILE,
 124                             formfile[1])))])
 125         elif filename is not None:
 126             with open(filename, 'rb') as stream:
 127                 postdata=stream.read()
 128             self._logger.verbose("Request POSTFIELDS: Binary content")
 129             curl_cmd.setopt(pycurl.POSTFIELDS, postdata)
 130
 131         if put_method:
 132             self._logger.info("Request METHOD: {} URL: {}".format("PUT",self._url + endpoint))
 133         elif patch_method:
 134             self._logger.info("Request METHOD: {} URL: {}".format("PATCH",self._url + endpoint))
 135         else:
 136             self._logger.info("Request METHOD: {} URL: {}".format("POST",self._url + endpoint))
 137         curl_cmd.perform()
 138         http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
 139         self._logger.info("Response HTTPCODE: {}".format(http_code))
 140         curl_cmd.close()
 141         self.check_http_response(http_code, data)
 142         if data.getvalue():
 143             data_text = data.getvalue().decode()
 144             self._logger.verbose("Response DATA: {}".format(data_text))
 145             return http_code, data_text
 146         else:
 147             return http_code, None
 148
 149     def post_cmd(self, endpoint='', postfields_dict=None,
 150                  formfile=None, filename=None,
 151                  skip_query_admin=False):
 152         self._logger.debug("")
 153         return self.send_cmd(endpoint=endpoint,
 154                              postfields_dict=postfields_dict,
 155                              formfile=formfile, filename=filename,
 156                              put_method=False, patch_method=False,
 157                              skip_query_admin=skip_query_admin)
 158
 159     def put_cmd(self, endpoint='', postfields_dict=None,
 160                 formfile=None, filename=None,
 161                 skip_query_admin=False):
 162         self._logger.debug("")
 163         return self.send_cmd(endpoint=endpoint,
 164                              postfields_dict=postfields_dict,
 165                              formfile=formfile, filename=filename,
 166                              put_method=True, patch_method=False,
 167                              skip_query_admin=skip_query_admin)
 168
 169     def patch_cmd(self, endpoint='', postfields_dict=None,
 170                 formfile=None, filename=None,
 171                 skip_query_admin=False):
 172         self._logger.debug("")
 173         return self.send_cmd(endpoint=endpoint,
 174                              postfields_dict=postfields_dict,
 175                              formfile=formfile, filename=filename,
 176                              put_method=False, patch_method=True,
 177                              skip_query_admin=skip_query_admin)
 178
 179     def get2_cmd(self, endpoint, skip_query_admin=False):
 180         self._logger.debug("")
 181         data = BytesIO()
 182         curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
 183         curl_cmd.setopt(pycurl.HTTPGET, 1)
 184         curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
 185         self._logger.info("Request METHOD: {} URL: {}".format("GET",self._url + endpoint))
 186         curl_cmd.perform()
 187         http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
 188         self._logger.info("Response HTTPCODE: {}".format(http_code))
 189         curl_cmd.close()
 190         self.check_http_response(http_code, data)
 191         if data.getvalue():
 192             data_text = data.getvalue().decode()
 193             self._logger.verbose("Response DATA: {}".format(data_text))
 194             return http_code, data_text
 195         return http_code, None
 196
 197     def check_http_response(self, http_code, data):
 198         if http_code >= 300:
 199             resp = ""
 200             if data.getvalue():
 201                 data_text = data.getvalue().decode()
 202                 self._logger.verbose("Response {} DATA: {}".format(http_code, data_text))
 203                 resp = ": " + data_text
 204             else:
 205                 self._logger.verbose("Response {}".format(http_code))
 206             if http_code == 404:
 207                 raise NotFound("Error {}{}".format(http_code, resp))
 208             raise OsmHttpException("Error {}{}".format(http_code, resp))
 209
 210     def set_query_admin(self, **kwargs):
 211         if 'all_projects' in kwargs:
 212             self._all_projects=kwargs['all_projects']
 213         if 'public' in kwargs:
 214             self._public=kwargs['public']
 215         self._default_query_admin = self._complete_default_query_admin()