5 from urllib import urlencode
10 from robot.libraries.BuiltIn import BuiltIn
13 class RequestsLibrary(object):
14 ROBOT_LIBRARY_SCOPE = 'Global'
17 self._cache = robot.utils.ConnectionCache('No sessions created')
18 self.builtin = BuiltIn()
20 def _utf8_urlencode(self, data):
21 if not type(data) is dict:
25 for k, v in data.iteritems():
26 utf8_data[k] = unicode(v).encode('utf-8')
27 return urlencode(utf8_data)
29 def create_session(self, alias, url, headers={}, cookies=None,
30 auth=None, timeout=None, proxies=None,
33 """ Create Session: create a HTTP session to a server
35 `url` Base url of the server
37 `alias` Robot Framework alias to identify the session
39 `headers` Dictionary of default headers
41 `auth` Dictionary of username & password for HTTP Basic Auth
43 `timeout` connection timeout
45 `proxies` proxy server url
47 `verify` set to True if Requests should verify the certificate
50 self.builtin.log('Creating session: %s' % alias, 'DEBUG')
51 auth = requests.auth.HTTPBasicAuth(*auth) if auth else None
52 s = session = requests.Session()
53 s.headers.update(headers)
54 s.auth = auth if auth else s.auth
55 s.proxies = proxies if proxies else s.proxies
57 s.verify = self.builtin.convert_to_boolean(verify)
59 # cant pass these into the Session anymore
60 self.timeout = timeout
61 self.cookies = cookies
67 self._cache.register(session, alias=alias)
70 def delete_all_sessions(self):
71 """Removes all the session objects"""
73 self._cache.empty_cache()
75 def to_json(self, content):
76 """Convert a string to a JSON object
78 `content` String content to convert into JSON
80 return json.loads(content)
82 def _get_url(self, session, uri):
83 """Helpere method to get the full url"""
86 slash = '' if uri.startswith('/') else '/'
87 url = "%s%s%s" % (session.url, slash, uri)
90 def get(self, alias, uri, headers=None):
91 """Send a GET request on the session object found using the given `alias`
93 `alias` that will be used to identify the Session object in the cache
95 `uri` to send the GET request to
97 `headers` a dictionary of headers to use with the request
100 "Use the Robot RequestsLibrary rather than this. See DatastoreCRUD.robot for examples",
103 session = self._cache.switch(alias)
104 resp = session.get(self._get_url(session, uri),
106 cookies=self.cookies, timeout=self.timeout)
108 # store the last response object
109 session.last_resp = resp
112 def post(self, alias, uri, data={}, headers=None, files={}):
113 """Send a POST request on the session object found using the given `alias`
115 `alias` that will be used to identify the Session object in the cache
117 `uri` to send the GET request to
119 `data` a dictionary of key-value pairs that will be urlencoded
120 and sent as POST data
121 or binary data that is sent as the raw body content
123 `headers` a dictionary of headers to use with the request
125 `files` a dictionary of file names containing file data to POST to the server
129 "Use the Robot RequestsLibrary rather than this. See DatastoreCRUD.robot for examples",
132 session = self._cache.switch(alias)
133 data = self._utf8_urlencode(data)
135 resp = session.post(self._get_url(session, uri),
136 data=data, headers=headers,
138 cookies=self.cookies, timeout=self.timeout)
140 # store the last response object
141 session.last_resp = resp
142 self.builtin.log("Post response: " + resp.content, 'DEBUG')
145 def postjson(self, alias, uri, data={}, headers=None, files={}):
146 """Send a POST request on the session object found using the given `alias`
148 `alias` that will be used to identify the Session object in the cache
150 `uri` to send the GET request to
152 `data` a dictionary of key-value pairs that will be urlencoded
153 and sent as POST data
154 or binary data that is sent as the raw body content
156 `headers` a dictionary of headers to use with the request
158 `files` a dictionary of file names containing file data to POST to the server
162 "Use the Robot RequestsLibrary rather than this. See DatastoreCRUD.robot for examples",
165 session = self._cache.switch(alias)
166 data = json.dumps(data)
168 resp = session.post(self._get_url(session, uri),
169 data=data, headers=headers,
171 cookies=self.cookies, timeout=self.timeout)
173 # store the last response object
174 session.last_resp = resp
175 self.builtin.log("Post response: " + resp.content, 'DEBUG')
178 def put(self, alias, uri, data=None, headers=None):
179 """Send a PUT request on the session object found using the given `alias`
181 `alias` that will be used to identify the Session object in the cache
183 `uri` to send the PUT request to
185 `headers` a dictionary of headers to use with the request
190 "Use the Robot RequestsLibrary rather than this. See DatastoreCRUD.robot for examples",
193 session = self._cache.switch(alias)
194 # data = self._utf8_urlencode(data)
195 data = json.dumps(data)
197 resp = session.put(self._get_url(session, uri),
198 data=data, headers=headers,
199 cookies=self.cookies, timeout=self.timeout)
201 self.builtin.log("PUT response: %s DEBUG" % resp.content)
203 # store the last response object
204 session.last_resp = resp
207 def put_xml(self, alias, uri, data=None, headers=None):
208 """Send a PUT_xml request on the session object found using the given `alias`
210 `alias` that will be used to identify the Session object in the cache
212 `uri` to send the PUT_xml request to
214 `headers` a dictionary of headers to use with the request
219 "Use the Robot RequestsLibrary rather than this. See DatastoreCRUD.robot for examples",
222 session = self._cache.switch(alias)
223 data = self._utf8_urlencode(data)
224 # data = json.dumps(data)
226 resp = session.put(self._get_url(session, uri),
227 data=data, headers=headers,
228 cookies=self.cookies, timeout=self.timeout)
230 self.builtin.log("PUT response: %s DEBUG" % resp.content)
232 # store the last response object
233 session.last_resp = resp
236 def delete(self, alias, uri, data=(), headers=None):
237 """Send a DELETE request on the session object found using the given `alias`
239 `alias` that will be used to identify the Session object in the cache
241 `uri` to send the DELETE request to
243 `headers` a dictionary of headers to use with the request
248 "Use the Robot RequestsLibrary rather than this. See DatastoreCRUD.robot for examples",
251 session = self._cache.switch(alias)
252 args = "?%s" % urlencode(data) if data else ''
253 resp = session.delete("%s%s" % (self._get_url(session, uri), args),
254 headers=headers, cookies=self.cookies,
255 timeout=self.timeout)
257 # store the last response object
258 session.last_resp = resp
261 def head(self, alias, uri, headers=None):
262 """Send a HEAD request on the session object found using the given `alias`
264 `alias` that will be used to identify the Session object in the cache
266 `uri` to send the HEAD request to
268 `headers` a dictionary of headers to use with the request
272 session = self._cache.switch(alias)
273 resp = session.head(self._get_url(session, uri), headers=headers,
274 cookies=self.cookies, timeout=self.timeout)
276 # store the last response object
277 session.last_resp = resp