osmclient/common/http.py

   1 # Copyright 2017 Sandvine
   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
  21
  22 class Http(object):
  23     def __init__(self, url, user="admin", password="admin"):
  24         self._url = url
  25         self._user = user
  26         self._password = password
  27         self._http_header = None
  28
  29     def set_http_header(self, header):
  30         self._http_header = header
  31
  32     def _get_curl_cmd(self, endpoint):
  33         curl_cmd = pycurl.Curl()
  34         curl_cmd.setopt(pycurl.URL, self._url + endpoint)
  35         curl_cmd.setopt(pycurl.SSL_VERIFYPEER, 0)
  36         curl_cmd.setopt(pycurl.SSL_VERIFYHOST, 0)
  37         curl_cmd.setopt(pycurl.USERPWD, "{}:{}".format(self._user, self._password))
  38         if self._http_header:
  39             curl_cmd.setopt(pycurl.HTTPHEADER, self._http_header)
  40         return curl_cmd
  41
  42     def get_cmd(self, endpoint):
  43
  44         data = BytesIO()
  45         curl_cmd = self._get_curl_cmd(endpoint)
  46         curl_cmd.setopt(pycurl.HTTPGET, 1)
  47         curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
  48         self._logger.info(
  49             "Request METHOD: {} URL: {}".format("GET", self._url + endpoint)
  50         )
  51         curl_cmd.perform()
  52         http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
  53         self._logger.info("Response HTTPCODE: {}".format(http_code))
  54         curl_cmd.close()
  55         if data.getvalue():
  56             self._logger.debug(
  57                 "Response DATA: {}".format(json.loads(data.getvalue().decode()))
  58             )
  59             return json.loads(data.getvalue().decode())
  60         return None
  61
  62     def delete_cmd(self, endpoint):
  63         data = BytesIO()
  64         curl_cmd = self._get_curl_cmd(endpoint)
  65         curl_cmd.setopt(pycurl.CUSTOMREQUEST, "DELETE")
  66         curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
  67         self._logger.info(
  68             "Request METHOD: {} URL: {}".format("DELETE", self._url + endpoint)
  69         )
  70         curl_cmd.perform()
  71         http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
  72         self._logger.info("Response HTTPCODE: {}".format(http_code))
  73         curl_cmd.close()
  74         if data.getvalue():
  75             self._logger.debug(
  76                 "Response DATA: {}".format(json.loads(data.getvalue().decode()))
  77             )
  78             return json.loads(data.getvalue().decode())
  79         return None
  80
  81     def post_cmd(
  82         self,
  83         endpoint="",
  84         postfields_dict=None,
  85         formfile=None,
  86     ):
  87         data = BytesIO()
  88         curl_cmd = self._get_curl_cmd(endpoint)
  89         curl_cmd.setopt(pycurl.POST, 1)
  90         curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
  91
  92         if postfields_dict is not None:
  93             jsondata = json.dumps(postfields_dict)
  94             curl_cmd.setopt(pycurl.POSTFIELDS, jsondata)
  95
  96         if formfile is not None:
  97             curl_cmd.setopt(
  98                 pycurl.HTTPPOST, [((formfile[0], (pycurl.FORM_FILE, formfile[1])))]
  99             )
 100
 101         self._logger.info(
 102             "Request METHOD: {} URL: {}".format("POST", self._url + endpoint)
 103         )
 104         curl_cmd.perform()
 105         http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
 106         self._logger.info("Response HTTPCODE: {}".format(http_code))
 107         curl_cmd.close()
 108         if data.getvalue():
 109             self._logger.debug(
 110                 "Response DATA: {}".format(json.loads(data.getvalue().decode()))
 111             )
 112             return json.loads(data.getvalue().decode())
 113         return None