Updated .gitignore: .idea
[osm/osmclient.git] / osmclient / sol005 / http.py
1 # Copyright 2018 Telefonica
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 from osmclient.common import http
21
22 class Http(http.Http):
23
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
30 def _get_curl_cmd(self, endpoint):
31 curl_cmd = pycurl.Curl()
32 #print self._url + endpoint
33 curl_cmd.setopt(pycurl.URL, self._url + endpoint)
34 curl_cmd.setopt(pycurl.SSL_VERIFYPEER, 0)
35 curl_cmd.setopt(pycurl.SSL_VERIFYHOST, 0)
36 if self._http_header:
37 curl_cmd.setopt(pycurl.HTTPHEADER, self._http_header)
38 return curl_cmd
39
40 def delete_cmd(self, endpoint):
41 data = BytesIO()
42 curl_cmd = self._get_curl_cmd(endpoint)
43 curl_cmd.setopt(pycurl.CUSTOMREQUEST, "DELETE")
44 curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
45 curl_cmd.perform()
46 http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
47 #print 'HTTP_CODE: {}'.format(http_code)
48 curl_cmd.close()
49 # TODO 202 accepted should be returned somehow
50 if data.getvalue():
51 return http_code, data.getvalue().decode()
52 else:
53 return http_code, None
54
55 def send_cmd(self, endpoint='', postfields_dict=None,
56 formfile=None, filename=None,
57 put_method=False, patch_method=False):
58 data = BytesIO()
59 curl_cmd = self._get_curl_cmd(endpoint)
60 if put_method:
61 curl_cmd.setopt(pycurl.CUSTOMREQUEST, "PUT")
62 elif patch_method:
63 curl_cmd.setopt(pycurl.CUSTOMREQUEST, "PATCH")
64 curl_cmd.setopt(pycurl.POST, 1)
65 curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
66
67 if postfields_dict is not None:
68 jsondata = json.dumps(postfields_dict)
69 curl_cmd.setopt(pycurl.POSTFIELDS, jsondata)
70 elif formfile is not None:
71 curl_cmd.setopt(
72 pycurl.HTTPPOST,
73 [((formfile[0],
74 (pycurl.FORM_FILE,
75 formfile[1])))])
76 elif filename is not None:
77 with open(filename, 'rb') as stream:
78 postdata=stream.read()
79 curl_cmd.setopt(pycurl.POSTFIELDS, postdata)
80
81 curl_cmd.perform()
82 http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
83 curl_cmd.close()
84 if data.getvalue():
85 return http_code, data.getvalue().decode()
86 else:
87 return http_code, None
88
89 def post_cmd(self, endpoint='', postfields_dict=None,
90 formfile=None, filename=None):
91 return self.send_cmd(endpoint=endpoint,
92 postfields_dict=postfields_dict,
93 formfile=formfile,
94 filename=filename,
95 put_method=False, patch_method=False)
96
97 def put_cmd(self, endpoint='', postfields_dict=None,
98 formfile=None, filename=None):
99 return self.send_cmd(endpoint=endpoint,
100 postfields_dict=postfields_dict,
101 formfile=formfile,
102 filename=filename,
103 put_method=True, patch_method=False)
104
105 def patch_cmd(self, endpoint='', postfields_dict=None,
106 formfile=None, filename=None):
107 return self.send_cmd(endpoint=endpoint,
108 postfields_dict=postfields_dict,
109 formfile=formfile,
110 filename=filename,
111 put_method=False, patch_method=True)
112
113 def get2_cmd(self, endpoint):
114 data = BytesIO()
115 curl_cmd = self._get_curl_cmd(endpoint)
116 curl_cmd.setopt(pycurl.HTTPGET, 1)
117 curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
118 curl_cmd.perform()
119 http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
120 curl_cmd.close()
121 if data.getvalue():
122 return http_code, data.getvalue().decode()
123 return http_code, None
124