dispatcher use separate sync/async, throttling

- use separate futures session to avoid blocking shorter
  api calls when async pool is exhausted
- add global throttle to avoid rate limits
- some DRY on header handling, comments + logging
- adjust former async handling of responses where needed
  e.g. remove resp.result()

planet/api/client.py

@@ -75,7 +75,7 @@ def login(self, identity, credentials):
         result = self.dispatcher.session.post(self._url('auth/login'), {
             'email': identity,
             'password': credentials
-        }).result()
+        })
         status = result.status_code
         if status == 400:
             raise APIException('invalid parameters, login process has changed')

planet/api/dispatch.py

@@ -12,10 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 import os
 import re
+import threading
+import time
 from requests_futures.sessions import FuturesSession
 from requests import Request
+from requests import Session
 from . utils import check_status
 from . models import Response
 from . exceptions import InvalidAPIKey
@@ -24,14 +28,41 @@
 
 USE_STRICT_SSL = not (os.getenv('DISABLE_STRICT_SSL', '').lower() == 'true')
 
+log = logging.getLogger(__name__)
+
+
+def _log_request(req):
+    log.info('%s %s %s %s', req.method, req.url, req.params, req.data)
+
 
 def _is_subdomain_of_tld(url1, url2):
     orig_host = urlparse(url1).hostname
     re_host = urlparse(url2).hostname
     return orig_host.split('.')[-2:] == re_host.split('.')[-2:]
 
 
-class RedirectSession(FuturesSession):
+class _Throttler(object):
+    '''A context manager that allows at most ops/sec
+    to avoid request throttling for all client operations.
+    '''
+
+    def __init__(self, ops=4):
+        self._cond = threading.Semaphore()
+        self._wait = 1./ops
+
+    def __enter__(self):
+        self._cond.acquire()
+        self._t = time.time()
+
+    def __exit__(self, *exc):
+        w = self._wait - (time.time() - self._t)
+        if w > 0:
+            time.sleep(w)
+        self._cond.release()
+        return False
+
+
+class RedirectSession(Session):
     '''This exists to override the existing behavior of requests that will
     strip Authorization headers from any redirect requests that resolve to a
     new domain. Instead, we'll keep headers if the redirect is a subdomain
@@ -54,39 +85,56 @@ def rebuild_auth(self, prepared_request, response):
                     )
 
 
+def _headers(request):
+    headers = {}
+    if request.data:
+        headers['Content-Type'] = 'application/json'
+    if request.auth:
+        headers.update({
+            'Authorization': 'api-key %s' % request.auth.value
+        })
+    else:
+        raise InvalidAPIKey('No API key provided')
+    return headers
+
+
 class RequestsDispatcher(object):
 
     def __init__(self, workers=4):
-        self.session = RedirectSession(max_workers=workers)
+        # general session for sync api calls
+        self.session = RedirectSession()
+        # the asyncpool is reserved for long-running async tasks
+        self._asyncpool = FuturesSession(
+            max_workers=workers,
+            session=self.session)
+        self._throttler = _Throttler()
 
     def response(self, request):
         return Response(request, self)
 
     def _dispatch_async(self, request, callback):
-        auth = request.auth
-        if not auth:
-            raise InvalidAPIKey('No API key provided')
-
-        def _auth_callback(req):
-            req.headers.update({
-                'Authorization': 'api-key %s' % auth.value
-            })
-            return req
-
-        headers = {}
-        if request.data:
-            headers['Content-Type'] = 'application/json'
-        return self.session.request(
-            request.method, request.url, data=request.data, headers=headers,
-            params=request.params, stream=True, background_callback=callback,
-            auth=_auth_callback, verify=USE_STRICT_SSL
-        )
+        with self._throttler:
+            _log_request(request)
+            return self._asyncpool.request(
+                request.method, request.url, data=request.data,
+                headers=_headers(request), params=request.params, stream=True,
+                background_callback=callback, verify=USE_STRICT_SSL
+            )
 
     def _dispatch(self, request, callback=None):
-        response = self._dispatch_async(request, callback).result()
-        check_status(response)
-        return response
-
+        with self._throttler:
+            _log_request(request)
+            t = time.time()
+            response = self.session.request(
+                request.method, request.url, data=request.data,
+                headers=_headers(request), params=request.params, stream=True,
+                verify=USE_STRICT_SSL
+            )
+            check_status(response)
+            log.info('request took %.03f', time.time() - t)
+            return response
+
+    # @todo delete me w/ v0 removal
     def dispatch_request(self, method, url, auth=None, params=None, data=None):
         headers = {}
         content_type = 'application/json'
@@ -96,4 +144,5 @@ def dispatch_request(self, method, url, auth=None, params=None, data=None):
                 'Content-Type': content_type
             })
         req = Request(method, url, params=params, data=data, headers=headers)
+        _log_request(req)
         return self.session.send(req.prepare(), verify=USE_STRICT_SSL)

tests/test_dispatch.py

@@ -26,14 +26,14 @@ def test_redirectsession_rebuilt_auth_called():
         m.get('http://newredirect.com', text='redirected!')
 
         # base assertion, works as intended
-        resp = session.get('http://redirect.com').result()
+        resp = session.get('http://redirect.com')
         assert resp.url == 'http://newredirect.com'
         assert resp.text == 'redirected!'
 
         # Authorization headers unpacked and URL is rewritten
         resp = session.get('http://redirect.com', headers={
             'Authorization': 'api-key foobar'
-        }).result()
+        })
         assert resp.url == 'http://newredirect.com/?api_key=foobar'
         assert resp.text == 'redirected!'
 
@@ -44,7 +44,7 @@ def test_redirectsession_rebuilt_auth_called():
         m.get('http://newredirect.com?param=yep', text='param!')
         resp = session.get('http://redirect.com?param=yep', headers={
             'Authorization': 'api-key foobar'
-        }).result()
+        })
         assert resp.url == 'http://newredirect.com/?param=yep&api_key=foobar'
         assert resp.text == 'param!'