4 from urllib import urlencode
8 from robot.libraries.BuiltIn import BuiltIn
11 class RequestsLibrary(object):
12 ROBOT_LIBRARY_SCOPE = 'Global'
15 self._cache = robot.utils.ConnectionCache('No sessions created')
16 self.builtin = BuiltIn()
18 def _utf8_urlencode(self, data):
19 if not type(data) is dict:
23 for k, v in data.iteritems():
24 utf8_data[k] = unicode(v).encode('utf-8')
25 return urlencode(utf8_data)
27 def create_session(self, alias, url, headers={}, cookies=None,
28 auth=None, timeout=None, proxies=None,
31 """ Create Session: create a HTTP session to a server
33 `url` Base url of the server
35 `alias` Robot Framework alias to identify the session
37 `headers` Dictionary of default headers
39 `auth` Dictionary of username & password for HTTP Basic Auth
41 `timeout` connection timeout
43 `proxies` proxy server url
45 `verify` set to True if Requests should verify the certificate
48 self.builtin.log('Creating session: %s' % alias, 'DEBUG')
49 auth = requests.auth.HTTPBasicAuth(*auth) if auth else None
50 s = session = requests.Session()
51 s.headers.update(headers)
52 s.auth = auth if auth else s.auth
53 s.proxies = proxies if proxies else s.proxies
55 s.verify = self.builtin.convert_to_boolean(verify)
57 # cant pass these into the Session anymore
58 self.timeout = timeout
59 self.cookies = cookies
65 self._cache.register(session, alias=alias)
68 def delete_all_sessions(self):
69 """ Removes all the session objects """
71 self._cache.empty_cache()
73 def to_json(self, content):
74 """ Convert a string to a JSON object
76 `content` String content to convert into JSON
78 return json.loads(content)
80 def _get_url(self, session, uri):
81 ''' Helpere method to get the full url
85 slash = '' if uri.startswith('/') else '/'
86 url = "%s%s%s" % (session.url, slash, uri)
89 def get(self, alias, uri, headers=None):
90 """ Send a GET request on the session object found using the
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 session = self._cache.switch(alias)
101 resp = session.get(self._get_url(session, uri),
103 cookies=self.cookies, timeout=self.timeout)
105 # store the last response object
106 session.last_resp = resp
109 def post(self, alias, uri, data={}, headers=None, files={}):
110 """ Send a POST request on the session object found using the
113 `alias` that will be used to identify the Session object in the cache
115 `uri` to send the GET request to
117 `data` a dictionary of key-value pairs that will be urlencoded
118 and sent as POST data
119 or binary data that is sent as the raw body content
121 `headers` a dictionary of headers to use with the request
123 `files` a dictionary of file names containing file data to POST to the server
126 session = self._cache.switch(alias)
127 data = self._utf8_urlencode(data)
129 resp = session.post(self._get_url(session, uri),
130 data=data, headers=headers,
132 cookies=self.cookies, timeout=self.timeout)
134 # store the last response object
135 session.last_resp = resp
136 self.builtin.log("Post response: " + resp.content, 'DEBUG')
139 def put(self, alias, uri, data=None, headers=None):
140 """ Send a PUT request on the session object found using the
143 `alias` that will be used to identify the Session object in the cache
145 `uri` to send the PUT request to
147 `headers` a dictionary of headers to use with the request
151 session = self._cache.switch(alias)
152 # data = json.dumps(self._utf8_urlencode(data))
153 data = json.dumps(data)
155 resp = session.put(self._get_url(session, uri),
156 data=data, headers=headers,
157 cookies=self.cookies, timeout=self.timeout)
159 self.builtin.log("PUT response: %s DEBUG" % resp.content)
161 # store the last response object
162 session.last_resp = resp
165 def delete(self, alias, uri, data=(), headers=None):
166 """ Send a DELETE request on the session object found using the
169 `alias` that will be used to identify the Session object in the cache
171 `uri` to send the DELETE request to
173 `headers` a dictionary of headers to use with the request
177 session = self._cache.switch(alias)
178 args = "?%s" % urlencode(data) if data else ''
179 resp = session.delete("%s%s" % (self._get_url(session, uri), args),
180 headers=headers, cookies=self.cookies,
181 timeout=self.timeout)
183 # store the last response object
184 session.last_resp = resp
187 def head(self, alias, uri, headers=None):
188 """ Send a HEAD request on the session object found using the
191 `alias` that will be used to identify the Session object in the cache
193 `uri` to send the HEAD request to
195 `headers` a dictionary of headers to use with the request
199 session = self._cache.switch(alias)
200 resp = session.head(self._get_url(session, uri), headers=headers,
201 cookies=self.cookies, timeout=self.timeout)
203 # store the last response object
204 session.last_resp = resp