codekingpro/portable-devtools

1import time2import logging3from authlib.common.urls import urlparse4from .errors import (5    MissingRequestTokenError,6    MissingTokenError,7)8from .sync_app import OAuth1Base, OAuth2Base9 10log = logging.getLogger(__name__)11 12__all__ = ['AsyncOAuth1Mixin', 'AsyncOAuth2Mixin']13 14 15class AsyncOAuth1Mixin(OAuth1Base):16    async def request(self, method, url, token=None, **kwargs):17        async with self._get_oauth_client() as session:18            return await _http_request(self, session, method, url, token, kwargs)19 20    async def create_authorization_url(self, redirect_uri=None, **kwargs):21        """Generate the authorization url and state for HTTP redirect.22 23        :param redirect_uri: Callback or redirect URI for authorization.24        :param kwargs: Extra parameters to include.25        :return: dict26        """27        if not self.authorize_url:28            raise RuntimeError('Missing "authorize_url" value')29 30        if self.authorize_params:31            kwargs.update(self.authorize_params)32 33        async with self._get_oauth_client() as client:34            client.redirect_uri = redirect_uri35            params = {}36            if self.request_token_params:37                params.update(self.request_token_params)38            request_token = await client.fetch_request_token(self.request_token_url, **params)39            log.debug(f'Fetch request token: {request_token!r}')40            url = client.create_authorization_url(self.authorize_url, **kwargs)41            state = request_token['oauth_token']42        return {'url': url, 'request_token': request_token, 'state': state}43 44    async def fetch_access_token(self, request_token=None, **kwargs):45        """Fetch access token in one step.46 47        :param request_token: A previous request token for OAuth 1.48        :param kwargs: Extra parameters to fetch access token.49        :return: A token dict.50        """51        async with self._get_oauth_client() as client:52            if request_token is None:53                raise MissingRequestTokenError()54            # merge request token with verifier55            token = {}56            token.update(request_token)57            token.update(kwargs)58            client.token = token59            params = self.access_token_params or {}60            token = await client.fetch_access_token(self.access_token_url, **params)61        return token62 63 64class AsyncOAuth2Mixin(OAuth2Base):65    async def _on_update_token(self, token, refresh_token=None, access_token=None):66        if self._update_token:67            await self._update_token(68                token,69                refresh_token=refresh_token,70                access_token=access_token,71            )72 73    async def load_server_metadata(self):74        if self._server_metadata_url and '_loaded_at' not in self.server_metadata:75            async with self.client_cls(**self.client_kwargs) as client:76                resp = await client.request('GET', self._server_metadata_url, withhold_token=True)77                resp.raise_for_status()78                metadata = resp.json()79                metadata['_loaded_at'] = time.time()80            self.server_metadata.update(metadata)81        return self.server_metadata82 83    async def request(self, method, url, token=None, **kwargs):84        metadata = await self.load_server_metadata()85        async with self._get_oauth_client(**metadata) as session:86            return await _http_request(self, session, method, url, token, kwargs)87 88    async def create_authorization_url(self, redirect_uri=None, **kwargs):89        """Generate the authorization url and state for HTTP redirect.90 91        :param redirect_uri: Callback or redirect URI for authorization.92        :param kwargs: Extra parameters to include.93        :return: dict94        """95        metadata = await self.load_server_metadata()96        authorization_endpoint = self.authorize_url or metadata.get('authorization_endpoint')97        if not authorization_endpoint:98            raise RuntimeError('Missing "authorize_url" value')99 100        if self.authorize_params:101            kwargs.update(self.authorize_params)102 103        async with self._get_oauth_client(**metadata) as client:104            client.redirect_uri = redirect_uri105            return self._create_oauth2_authorization_url(106                client, authorization_endpoint, **kwargs)107 108    async def fetch_access_token(self, redirect_uri=None,  **kwargs):109        """Fetch access token in the final step.110 111        :param redirect_uri: Callback or Redirect URI that is used in112                             previous :meth:`authorize_redirect`.113        :param kwargs: Extra parameters to fetch access token.114        :return: A token dict.115        """116        metadata = await self.load_server_metadata()117        token_endpoint = self.access_token_url or metadata.get('token_endpoint')118        async with self._get_oauth_client(**metadata) as client:119            if redirect_uri is not None:120                client.redirect_uri = redirect_uri121            params = {}122            if self.access_token_params:123                params.update(self.access_token_params)124            params.update(kwargs)125            token = await client.fetch_token(token_endpoint, **params)126        return token127 128 129async def _http_request(ctx, session, method, url, token, kwargs):130    request = kwargs.pop('request', None)131    withhold_token = kwargs.get('withhold_token')132    if ctx.api_base_url and not url.startswith(('https://', 'http://')):133        url = urlparse.urljoin(ctx.api_base_url, url)134 135    if withhold_token:136        return await session.request(method, url, **kwargs)137 138    if token is None and ctx._fetch_token and request:139        token = await ctx._fetch_token(request)140    if token is None:141        raise MissingTokenError()142 143    session.token = token144    return await session.request(method, url, **kwargs)145