Source code for pulsar.apps.http.oauth

try:
    from oauthlib import oauth1, oauth2
except ImportError:     # pragma    nocover
    oauth1 = None
    oauth2 = None

from pulsar import ImproperlyConfigured
from pulsar.utils.structures import mapping_iterator

from . import auth


[docs]class OAuth1(auth.Auth):
    '''Add OAuth1 authentication to pulsar :class:`.HttpClient`
    '''
    available = bool(oauth1)

    def __init__(self, client_id=None, client=None, **kw):
        if oauth1 is None:  # pragma    nocover
            raise ImproperlyConfigured('%s requires oauthlib' %
                                       self.__class__.__name__)
        self._client = client or oauth1.Client(client_id, **kw)

    def __call__(self, response, exc=None):
        r = response.request
        url, headers, _ = self._client.sign(
            r.url, r.method, r.body, r.headers)
        for key, value in mapping_iterator(headers):
            r.add_header(key, value)
        r.url = url


[docs]class OAuth2(auth.Auth):
    """Add OAuth2 authentication to pulsar :class:`.HttpClient`
    """
    available = bool(oauth2)

    def __init__(self, client_id=None, client=None, **kw):
        if oauth2 is None:  # pragma    nocover
            raise ImproperlyConfigured('%s requires oauthlib' %
                                       self.__class__.__name__)
        self.client = client or oauth2.WebApplicationClient(client_id, **kw)

    def __call__(self, response, exc=None):
        r = response.request
        url, headers, _ = self.client.add_token(
            r.url, http_method=r.method, body=r.body, headers=r.headers)
        for key, value in mapping_iterator(headers):
            r.add_header(key, value)
        r.url = url