Merge pull request #48 from cpaillet/master

Bump to v1.2.4: Add timeout for request

CHANGELOG.rst

@@ -1,6 +1,10 @@
 Change log
 ==========
 
+1.2.4
+-----
+* feature: aio: allow setting timeout by request
+
 1.2.3
 -----
 * feature: base: ensure return format of json callback is more consistent

consul/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '1.2.3'
+__version__ = '1.2.4'
 
 from consul.std import Consul
 

consul/aio.py

@@ -27,29 +27,33 @@ def __init__(self, *args, loop=None, connections_limit=None,
         self._session = aiohttp.ClientSession(connector=connector,
                                               **session_kwargs)
 
-    async def _request(self, callback, method, uri, data=None):
-        resp = await self._session.request(method, uri, data=data)
+    async def _request(self, callback, method, uri, data=None, connections_timeout=None):
+        session_kwargs = {}
+        if connections_timeout:
+            timeout = aiohttp.ClientTimeout(total=connections_timeout)
+            session_kwargs['timeout'] = timeout
+        resp = await self._session.request(method, uri, data=data, **session_kwargs)
         body = await resp.text(encoding='utf-8')
         if resp.status == 599:
             raise base.Timeout
         r = base.Response(resp.status, resp.headers, body)
         return callback(r)
 
-    def get(self, callback, path, params=None):
+    def get(self, callback, path, params=None, connections_timeout=None):
         uri = self.uri(path, params)
-        return self._request(callback, 'GET', uri)
+        return self._request(callback, 'GET', uri, connections_timeout=connections_timeout)
 
-    def put(self, callback, path, params=None, data=''):
+    def put(self, callback, path, params=None, data='', connections_timeout=None):
         uri = self.uri(path, params)
-        return self._request(callback, 'PUT', uri, data=data)
+        return self._request(callback, 'PUT', uri, data=data, connections_timeout=connections_timeout)
 
-    def delete(self, callback, path, params=None):
+    def delete(self, callback, path, params=None, connections_timeout=None):
         uri = self.uri(path, params)
-        return self._request(callback, 'DELETE', uri)
+        return self._request(callback, 'DELETE', uri, connections_timeout=connections_timeout)
 
-    def post(self, callback, path, params=None, data=''):
+    def post(self, callback, path, params=None, data='', connections_timeout=None):
         uri = self.uri(path, params)
-        return self._request(callback, 'POST', uri, data=data)
+        return self._request(callback, 'POST', uri, data=data, connections_timeout=connections_timeout)
 
     def close(self):
         return self._session.close()

consul/base.py

@@ -494,7 +494,8 @@ def get(
                 consistency=None,
                 keys=False,
                 separator=None,
-                dc=None):
+                dc=None,
+                connections_timeout=None):
             """
             Returns a tuple of (*index*, *value[s]*)
 
@@ -564,10 +565,13 @@ def get(
                 decode = 'Value'
             if not recurse and not keys:
                 one = True
+            http_kwargs = {}
+            if connections_timeout:
+                http_kwargs['connections_timeout'] = connections_timeout
             return self.agent.http.get(
                 CB.json(index=True, decode=decode, one=one),
                 '/v1/kv/%s' % key,
-                params=params)
+                params=params,**http_kwargs)
 
         def put(
                 self,
@@ -578,7 +582,8 @@ def put(
                 acquire=None,
                 release=None,
                 token=None,
-                dc=None):
+                dc=None,
+                connections_timeout=None):
             """
             Sets *key* to the given *value*.
 
@@ -632,10 +637,13 @@ def put(
             dc = dc or self.agent.dc
             if dc:
                 params.append(('dc', dc))
+            http_kwargs = {}
+            if connections_timeout:
+                http_kwargs['connections_timeout'] = connections_timeout
             return self.agent.http.put(
-                CB.json(), '/v1/kv/%s' % key, params=params, data=value)
+                CB.json(), '/v1/kv/%s' % key, params=params, data=value, **http_kwargs)
 
-        def delete(self, key, recurse=None, cas=None, token=None, dc=None):
+        def delete(self, key, recurse=None, cas=None, token=None, dc=None, connections_timeout=None):
             """
             Deletes a single key or if *recurse* is True, all keys sharing a
             prefix.
@@ -668,9 +676,11 @@ def delete(self, key, recurse=None, cas=None, token=None, dc=None):
             dc = dc or self.agent.dc
             if dc:
                 params.append(('dc', dc))
-
+            http_kwargs = {}
+            if connections_timeout:
+                http_kwargs['connections_timeout'] = connections_timeout
             return self.agent.http.delete(
-                CB.json(), '/v1/kv/%s' % key, params=params)
+                CB.json(), '/v1/kv/%s' % key, params=params, **http_kwargs)
 
     class Txn:
         """