Source code for aiomixcloud.auth

"""
API access authorization
~~~~~~~~~~~~~~~~~~~~~~~~

This module contains the class for Mixcloud API OAuth authorization.
Specifically:

    - :class:`MixcloudOAuth`, producing authorization URLs and
      trading OAuth codes for access tokens.
"""

import aiohttp
import yarl

from aiomixcloud.constants import OAUTH_ROOT
from aiomixcloud.exceptions import MixcloudOAuthError


[docs]class MixcloudOAuth: """Mixcloud OAuth authorization By having :attr:`client_id` and :attr:`redirect_uri` set, a :class:`MixcloudOAuth` object provides :attr:`authorization_url`, a URL to forward the end user to, where they will be able to "allow the application access to their data". It also provides the :meth:`access_token` method, which trades an OAuth code for an access token (requiring :attr:`client_secret` as well as the previously mentionted attributes). """ #: Default Mixcloud OAuth root URL oauth_root = OAUTH_ROOT def __init__(self, oauth_root=oauth_root, *, client_id=None, client_secret=None, redirect_uri=None, raise_exceptions=None, mixcloud=None): """Store instance attributes.""" #: Base URL for OAuth-related requests self._oauth_root = oauth_root #: Client ID, provided by Mixcloud self.client_id = client_id #: Client secret, provided by Mixcloud self.client_secret = client_secret #: Redirect URI, chosen by the developer self.redirect_uri = redirect_uri #: Whether to raise an exception when API responds #: with an error message. If not specified, use the respective #: setting of the :attr:`mixcloud` attribute. If that #: attribute is not specified, default to ``False``. self._raise_exceptions = raise_exceptions #: The :class:`~aiomixcloud.core.Mixcloud` object whose session #: will be used to make the request from. If ``None``, a new #: session will be created for the access token request. self.mixcloud = mixcloud def _check(self): """Check that :attr:`client ID <client_id>` and :attr:`redirect URI <redirect_uri>` have been set. """ assert self.client_id is not None, 'client_id must be set' assert self.redirect_uri is not None, 'redirect_uri must be set' def _build_url(self, segment): """Return a :class:`~yarl.URL` consisting of :attr:`OAuth root <_oauth_root>`, followed by `segment`. """ return yarl.URL(self._oauth_root) / segment @property def authorization_url(self): """Return authorization URL.""" self._check() params = {'client_id': self.client_id, 'redirect_uri': self.redirect_uri} url = self._build_url('authorize') final_url = url.with_query(params) return str(final_url)
[docs] async def access_token(self, code): """Send OAuth `code` to server and get an access token. If fail raise :class:`~aiomixcloud.exceptions.MixcloudOAuthError` in case this is the setting (:attr:`self._raise_exceptions <_raise_exceptions>` or :attr:`self.mixcloud._raise_exceptions`), otherwise return ``None``. """ self._check() assert self.client_secret is not None, 'client_secret must be set' params = {'client_id': self.client_id, 'redirect_uri': self.redirect_uri, 'client_secret': self.client_secret, 'code': code} url = self._build_url('access_token') if self.mixcloud is None: # No Mixcloud instance stored, start a new session. session = aiohttp.ClientSession() else: session = self.mixcloud._session async with session.get(url, params=params) as response: data = await response.json() # If started a new session, close it. if self.mixcloud is None: await session.close() try: return data['access_token'] except KeyError: if self._raise_exceptions is None: # Own setting not specified. If a Mixcloud instance # is stored, act according to its setting. if (self.mixcloud is not None and self.mixcloud._raise_exceptions): raise MixcloudOAuthError(data) from None elif self._raise_exceptions: # Own setting specified and dictates to # raise exception. raise MixcloudOAuthError(data) from None # No setting to raise exception. return None