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