- with open(filename, 'r') as stream:
- postdata=stream.read()
- curl_cmd.setopt(pycurl.POSTFIELDS, postdata)
-
- curl_cmd.perform()
- http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
- curl_cmd.close()
- if http_code not in (200, 201, 202, 204):
- raise ClientException(data.getvalue().decode())
- if postfields_dict is not None:
- if data.getvalue():
- return json.loads(data.getvalue().decode())
- return None
- elif formfile is not None:
- if data.getvalue():
- return yaml.safe_load(data.getvalue().decode())
- return None
- elif filename is not None:
- if data.getvalue():
- return yaml.safe_load(data.getvalue().decode())
- return None
- return None
-
- def post_cmd(self, endpoint='', postfields_dict=None,
- formfile=None, filename=None):
- return self.send_cmd(endpoint=endpoint,
- postfields_dict=postfields_dict,
- formfile=formfile,
- filename=filename,
- put_method=False)
-
- def put_cmd(self, endpoint='', postfields_dict=None,
- formfile=None, filename=None):
- return self.send_cmd(endpoint=endpoint,
- postfields_dict=postfields_dict,
- formfile=formfile,
- filename=filename,
- put_method=True)
-
- def get2_cmd(self, endpoint):
- data = BytesIO()
- curl_cmd = self._get_curl_cmd(endpoint)
- curl_cmd.setopt(pycurl.HTTPGET, 1)
- curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
- curl_cmd.perform()
- curl_cmd.close()
- return data.getvalue()
+ with open(filename, "rb") as stream:
+ postdata = stream.read()
+ self._logger.verbose("Request POSTFIELDS: Binary content")
+ requests_cmd.data = postdata
+
+ if put_method:
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("PUT", self._url + endpoint)
+ )
+ elif patch_method:
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("PATCH", self._url + endpoint)
+ )
+ else:
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("POST", self._url + endpoint)
+ )
+ requests_cmd = requests_cmd.prepare()
+ resp = session_cmd.send(
+ requests_cmd, verify=False, timeout=self.CONNECT_TIMEOUT
+ )
+ http_code = resp.status_code
+ self._logger.info("Response HTTPCODE: {}".format(http_code))
+ data = resp.content
+ self.check_http_response(http_code, data)
+ session_cmd.close()
+ if data:
+ data_text = data.decode()
+ self._logger.verbose("Response DATA: {}".format(data_text))
+ return http_code, data_text
+ return http_code, None
+
+ def post_cmd(
+ self,
+ endpoint="",
+ postfields_dict=None,
+ formfile=None,
+ filename=None,
+ skip_query_admin=False,
+ ):
+ self._logger.debug("")
+ return self.send_cmd(
+ endpoint=endpoint,
+ postfields_dict=postfields_dict,
+ formfile=formfile,
+ filename=filename,
+ put_method=False,
+ patch_method=False,
+ skip_query_admin=skip_query_admin,
+ )
+
+ def put_cmd(
+ self,
+ endpoint="",
+ postfields_dict=None,
+ formfile=None,
+ filename=None,
+ skip_query_admin=False,
+ ):
+ self._logger.debug("")
+ return self.send_cmd(
+ endpoint=endpoint,
+ postfields_dict=postfields_dict,
+ formfile=formfile,
+ filename=filename,
+ put_method=True,
+ patch_method=False,
+ skip_query_admin=skip_query_admin,
+ )
+
+ def patch_cmd(
+ self,
+ endpoint="",
+ postfields_dict=None,
+ formfile=None,
+ filename=None,
+ skip_query_admin=False,
+ ):
+ self._logger.debug("")
+ return self.send_cmd(
+ endpoint=endpoint,
+ postfields_dict=postfields_dict,
+ formfile=formfile,
+ filename=filename,
+ put_method=False,
+ patch_method=True,
+ skip_query_admin=skip_query_admin,
+ )
+
+ def get2_cmd(self, endpoint, skip_query_admin=False):
+ session_cmd = requests.Session()
+ self._logger.debug("")
+ requests_cmd = self._get_requests_cmd(endpoint, skip_query_admin)
+ requests_cmd.method = "GET"
+ self._logger.info(
+ "Request METHOD: {} URL: {}".format("GET", self._url + endpoint)
+ )
+ requests_cmd = requests_cmd.prepare()
+ resp = session_cmd.send(
+ requests_cmd, verify=False, timeout=self.CONNECT_TIMEOUT
+ )
+ http_code = resp.status_code
+ self._logger.info("Response HTTPCODE: {}".format(http_code))
+ data = resp.content
+ session_cmd.close()
+ self.check_http_response(http_code, data)
+ if data:
+ data_text = data.decode()
+ self._logger.verbose("Response DATA: {}".format(data_text))
+ return http_code, data_text
+ return http_code, None
+
+ def check_http_response(self, http_code, data):
+ if http_code >= 300:
+ resp = ""
+ if data:
+ data_text = data.decode()
+ self._logger.verbose(
+ "Response {} DATA: {}".format(http_code, data_text)
+ )
+ resp = ": " + data_text
+ else:
+ self._logger.verbose("Response {}".format(http_code))
+ if http_code == 404:
+ raise NotFound("Error {}{}".format(http_code, resp))
+ raise OsmHttpException("Error {}{}".format(http_code, resp))