class Http(object):
-
- def __init__(self, url, user='admin', password='admin'):
+ def __init__(self, url, user="admin", password="admin"):
self._url = url
self._user = user
self._password = password
curl_cmd.setopt(pycurl.URL, self._url + endpoint)
curl_cmd.setopt(pycurl.SSL_VERIFYPEER, 0)
curl_cmd.setopt(pycurl.SSL_VERIFYHOST, 0)
- curl_cmd.setopt(
- pycurl.USERPWD,
- '{}:{}'.format(
- self._user,
- self._password))
+ curl_cmd.setopt(pycurl.USERPWD, "{}:{}".format(self._user, self._password))
if self._http_header:
curl_cmd.setopt(pycurl.HTTPHEADER, self._http_header)
return curl_cmd
curl_cmd = self._get_curl_cmd(endpoint)
curl_cmd.setopt(pycurl.HTTPGET, 1)
curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
- self._logger.info("Request METHOD: {} URL: {}".format("GET",self._url + endpoint))
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("GET", self._url + endpoint)
+ )
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
curl_cmd.close()
if data.getvalue():
- self._logger.debug("Response DATA: {}".format(json.loads(data.getvalue().decode())))
+ self._logger.debug(
+ "Response DATA: {}".format(json.loads(data.getvalue().decode()))
+ )
return json.loads(data.getvalue().decode())
return None
curl_cmd = self._get_curl_cmd(endpoint)
curl_cmd.setopt(pycurl.CUSTOMREQUEST, "DELETE")
curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
- self._logger.info("Request METHOD: {} URL: {}".format("DELETE",self._url + endpoint))
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("DELETE", self._url + endpoint)
+ )
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
curl_cmd.close()
if data.getvalue():
- self._logger.debug("Response DATA: {}".format(json.loads(data.getvalue().decode())))
+ self._logger.debug(
+ "Response DATA: {}".format(json.loads(data.getvalue().decode()))
+ )
return json.loads(data.getvalue().decode())
return None
- def post_cmd(self, endpoint='', postfields_dict=None, formfile=None, ):
+ def post_cmd(
+ self,
+ endpoint="",
+ postfields_dict=None,
+ formfile=None,
+ ):
data = BytesIO()
curl_cmd = self._get_curl_cmd(endpoint)
curl_cmd.setopt(pycurl.POST, 1)
if formfile is not None:
curl_cmd.setopt(
- pycurl.HTTPPOST,
- [((formfile[0],
- (pycurl.FORM_FILE,
- formfile[1])))])
+ pycurl.HTTPPOST, [((formfile[0], (pycurl.FORM_FILE, formfile[1])))]
+ )
- self._logger.info("Request METHOD: {} URL: {}".format("POST",self._url + endpoint))
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("POST", self._url + endpoint)
+ )
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
curl_cmd.close()
if data.getvalue():
- self._logger.debug("Response DATA: {}".format(json.loads(data.getvalue().decode())))
+ self._logger.debug(
+ "Response DATA: {}".format(json.loads(data.getvalue().decode()))
+ )
return json.loads(data.getvalue().decode())
return None