Move client session objects to new module

Author: smithsz

src/cloudant/_common_util.py

@@ -17,14 +17,12 @@
 throughout the library.
 """
 
-import os
 import sys
 import platform
 from collections import Sequence
 import json
-from requests import RequestException, Session
 
-from ._2to3 import LONGTYPE, STRTYPE, NONETYPE, UNITYPE, iteritems_, url_join
+from ._2to3 import LONGTYPE, STRTYPE, NONETYPE, UNITYPE, iteritems_
 from .error import CloudantArgumentError, CloudantException
 
 # Library Constants
@@ -290,211 +288,6 @@ def __new__(cls, code):
         return str.__new__(cls, code)
 
 
-class ClientSession(Session):
-    """
-    This class extends Session and provides a default timeout.
-    """
-
-    def __init__(self, **kwargs):
-        super(ClientSession, self).__init__()
-        self._timeout = kwargs.get('timeout', None)
-
-    def request(self, method, url, **kwargs):  # pylint: disable=W0221
-        """
-        Overrides ``requests.Session.request`` to set the timeout.
-        """
-        resp = super(ClientSession, self).request(
-            method, url, timeout=self._timeout, **kwargs)
-
-        return resp
-
-
-class CookieSession(ClientSession):
-    """
-    This class extends ClientSession and provides cookie authentication.
-    """
-
-    def __init__(self, username, password, server_url, **kwargs):
-        super(CookieSession, self).__init__(**kwargs)
-        self._username = username
-        self._password = password
-        self._auto_renew = kwargs.get('auto_renew', False)
-        self._session_url = url_join(server_url, '_session')
-
-    def info(self):
-        """
-        Get cookie based login user information.
-        """
-        resp = self.get(self._session_url)
-        resp.raise_for_status()
-
-        return resp.json()
-
-    def login(self):
-        """
-        Perform cookie based user login.
-        """
-        resp = super(CookieSession, self).request(
-            'POST',
-            self._session_url,
-            data={'name': self._username, 'password': self._password},
-        )
-        resp.raise_for_status()
-
-    def logout(self):
-        """
-        Logout cookie based user.
-        """
-        resp = super(CookieSession, self).request('DELETE', self._session_url)
-        resp.raise_for_status()
-
-    def request(self, method, url, **kwargs):  # pylint: disable=W0221
-        """
-        Overrides ``requests.Session.request`` to renew the cookie and then
-        retry the original request (if required).
-        """
-        resp = super(CookieSession, self).request(method, url, **kwargs)
-
-        if not self._auto_renew:
-            return resp
-
-        is_expired = any((
-            resp.status_code == 403 and
-            resp.json().get('error') == 'credentials_expired',
-            resp.status_code == 401
-        ))
-
-        if is_expired:
-            self.login()
-            resp = super(CookieSession, self).request(method, url, **kwargs)
-
-        return resp
-
-    def set_credentials(self, username, password):
-        """
-        Set a new username and password.
-
-        :param str username: New username.
-        :param str password: New password.
-        """
-        if username is not None:
-            self._username = username
-
-        if password is not None:
-            self._password = password
-
-
-class IAMSession(ClientSession):
-    """
-    This class extends ClientSession and provides IAM authentication.
-    """
-
-    def __init__(self, api_key, server_url, **kwargs):
-        super(IAMSession, self).__init__(**kwargs)
-        self._api_key = api_key
-        self._auto_renew = kwargs.get('auto_renew', False)
-        self._session_url = url_join(server_url, '_iam_session')
-        self._token_url = os.environ.get(
-            'IAM_TOKEN_URL', 'https://iam.bluemix.net/oidc/token')
-
-    def info(self):
-        """
-        Get IAM cookie based login user information.
-        """
-        resp = self.get(self._session_url)
-        resp.raise_for_status()
-
-        return resp.json()
-
-    def login(self):
-        """
-        Perform IAM cookie based user login.
-        """
-        access_token = self._get_access_token()
-        try:
-            super(IAMSession, self).request(
-                'POST',
-                self._session_url,
-                headers={'Content-Type': 'application/json'},
-                data=json.dumps({'access_token': access_token})
-            ).raise_for_status()
-
-        except RequestException:
-            raise CloudantException(
-                'Failed to exchange IAM token with Cloudant')
-
-    def logout(self):
-        """
-        Logout IAM cookie based user.
-        """
-        self.cookies.clear()
-
-    def request(self, method, url, **kwargs):  # pylint: disable=W0221
-        """
-        Overrides ``requests.Session.request`` to renew the IAM cookie
-        and then retry the original request (if required).
-        """
-        # The CookieJar API prevents callers from getting an individual Cookie
-        # object by name.
-        # We are forced to use the only exposed method of discarding expired
-        # cookies from the CookieJar. Internally this involves iterating over
-        # the entire CookieJar and calling `.is_expired()` on each Cookie
-        # object.
-        self.cookies.clear_expired_cookies()
-
-        if self._auto_renew and 'IAMSession' not in self.cookies.keys():
-            self.login()
-
-        resp = super(IAMSession, self).request(method, url, **kwargs)
-
-        if not self._auto_renew:
-            return resp
-
-        if resp.status_code == 401:
-            self.login()
-            resp = super(IAMSession, self).request(method, url, **kwargs)
-
-        return resp
-
-    def set_credentials(self, username, api_key):
-        """
-        Set a new IAM API key.
-
-        :param str username: Username parameter is unused.
-        :param str api_key: New IAM API key.
-        """
-        if api_key is not None:
-            self._api_key = api_key
-
-    def _get_access_token(self):
-        """
-        Get IAM access token using API key.
-        """
-        err = 'Failed to contact IAM token service'
-        try:
-            resp = super(IAMSession, self).request(
-                'POST',
-                self._token_url,
-                auth=('bx', 'bx'),  # required for user API keys
-                headers={'Accepts': 'application/json'},
-                data={
-                    'grant_type': 'urn:ibm:params:oauth:grant-type:apikey',
-                    'response_type': 'cloud_iam',
-                    'apikey': self._api_key
-                }
-            )
-            err = resp.json().get('errorMessage', err)
-            resp.raise_for_status()
-
-            return resp.json()['access_token']
-
-        except KeyError:
-            raise CloudantException('Invalid response from IAM token service')
-
-        except RequestException:
-            raise CloudantException(err)
-
-
 class CloudFoundryService(object):
     """ Manages Cloud Foundry service configuration. """
 

src/cloudant/client.py

@@ -29,10 +29,10 @@
 from ._common_util import (
     USER_AGENT,
     append_response_error_content,
-    ClientSession,
     CloudFoundryService,
-    CookieSession,
-    IAMSession)
+    )
+from client_session import ClientSession, CookieSession, IAMSession
+
 
 class CouchDB(dict):
     """