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 import requests
  18 import urllib3
  19 from requests.auth import HTTPBasicAuth
  20 import json
  21 import logging
  22
  23 urllib3.disable_warnings()
  24
  25
  26 class Http(object):
  27     def __init__(self, url, user="admin", password="admin"):
  28         self._url = url
  29         self._user = user
  30         self._password = password
  31         self._http_header = None
  32         self._logger = logging.getLogger("osmclient")
  33
  34     def set_http_header(self, header):
  35         self._http_header = header
  36
  37     def _get_requests_cmd(self, endpoint):
  38         requests_cmd = requests.Request()
  39         requests_cmd.url = self._url + endpoint
  40         requests_cmd.auth = HTTPBasicAuth(self._user, self._password)
  41         if self._http_header:
  42             requests_cmd.headers = self._http_header
  43         return requests_cmd
  44
  45     def get_cmd(self, endpoint):
  46         session_cmd = requests.Session()
  47         requests_cmd = self._get_requests_cmd(endpoint)
  48         requests_cmd.method = "GET"
  49         self._logger.info(
  50             "Request METHOD: {} URL: {}".format("GET", self._url + endpoint)
  51         )
  52         requests_cmd = requests_cmd.prepare()
  53         resp = session_cmd.send(requests_cmd, verify=False)
  54         http_code = resp.status_code
  55         self._logger.info("Response HTTPCODE: {}".format(http_code))
  56         data = resp.content
  57         session_cmd.close()
  58         if data:
  59             self._logger.debug("Response DATA: {}".format(json.loads(data.decode())))
  60             return json.loads(data.decode())
  61         return None
  62
  63     def delete_cmd(self, endpoint):
  64         session_cmd = requests.Session()
  65         requests_cmd = self._get_requests_cmd(endpoint)
  66         requests_cmd.method = "DELETE"
  67         self._logger.info(
  68             "Request METHOD: {} URL: {}".format("DELETE", self._url + endpoint)
  69         )
  70         requests_cmd = requests_cmd.prepare()
  71         resp = session_cmd.send(requests_cmd, verify=False)
  72         http_code = resp.status_code
  73         self._logger.info("Response HTTPCODE: {}".format(http_code))
  74         data = resp.content
  75         session_cmd.close()
  76         if data:
  77             self._logger.debug("Response DATA: {}".format(json.loads(data.decode())))
  78             return json.loads(data.decode())
  79         return None
  80
  81     def post_cmd(
  82         self,
  83         endpoint="",
  84         postfields_dict=None,
  85         formfile=None,
  86     ):
  87         session_cmd = requests.Session()
  88         requests_cmd = self._get_requests_cmd(endpoint)
  89         requests_cmd.method = "POST"
  90
  91         if postfields_dict is not None:
  92             requests_cmd.json = json.dumps(postfields_dict)
  93
  94         if formfile is not None:
  95             requests_cmd.files = {formfile[0]: formfile[1]}
  96
  97         self._logger.info(
  98             "Request METHOD: {} URL: {}".format("POST", self._url + endpoint)
  99         )
 100         requests_cmd = requests_cmd.prepare()
 101         resp = session_cmd.send(requests_cmd, verify=False)
 102         http_code = resp.status_code
 103         self._logger.info("Response HTTPCODE: {}".format(http_code))
 104         data = resp.content
 105         session_cmd.close()
 106         if data:
 107             self._logger.debug("Response DATA: {}".format(json.loads(data.decode())))
 108             return json.loads(data.decode())
 109         return None