codekingpro/portable-devtools

1import typing2import httpx3from httpx import Auth, Request, Response4from authlib.oauth1 import (5    SIGNATURE_HMAC_SHA1,6    SIGNATURE_TYPE_HEADER,7)8from authlib.common.encoding import to_unicode9from authlib.oauth1 import ClientAuth10from authlib.oauth1.client import OAuth1Client as _OAuth1Client11from .utils import build_request, extract_client_kwargs12from ..base_client import OAuthError13 14 15class OAuth1Auth(Auth, ClientAuth):16    """Signs the httpx request using OAuth 1 (RFC5849)"""17    requires_request_body = True18 19    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:20        url, headers, body = self.prepare(21            request.method, str(request.url), request.headers, request.content)22        headers['Content-Length'] = str(len(body))23        yield build_request(url=url, headers=headers, body=body, initial_request=request)24 25 26class AsyncOAuth1Client(_OAuth1Client, httpx.AsyncClient):27    auth_class = OAuth1Auth28 29    def __init__(self, client_id, client_secret=None,30                 token=None, token_secret=None,31                 redirect_uri=None, rsa_key=None, verifier=None,32                 signature_method=SIGNATURE_HMAC_SHA1,33                 signature_type=SIGNATURE_TYPE_HEADER,34                 force_include_body=False, **kwargs):35 36        _client_kwargs = extract_client_kwargs(kwargs)37        httpx.AsyncClient.__init__(self, **_client_kwargs)38 39        _OAuth1Client.__init__(40            self, None,41            client_id=client_id, client_secret=client_secret,42            token=token, token_secret=token_secret,43            redirect_uri=redirect_uri, rsa_key=rsa_key, verifier=verifier,44            signature_method=signature_method, signature_type=signature_type,45            force_include_body=force_include_body, **kwargs)46 47    async def fetch_access_token(self, url, verifier=None, **kwargs):48        """Method for fetching an access token from the token endpoint.49 50        This is the final step in the OAuth 1 workflow. An access token is51        obtained using all previously obtained credentials, including the52        verifier from the authorization step.53 54        :param url: Access Token endpoint.55        :param verifier: A verifier string to prove authorization was granted.56        :param kwargs: Extra parameters to include for fetching access token.57        :return: A token dict.58        """59        if verifier:60            self.auth.verifier = verifier61        if not self.auth.verifier:62            self.handle_error('missing_verifier', 'Missing "verifier" value')63        token = await self._fetch_token(url, **kwargs)64        self.auth.verifier = None65        return token66 67    async def _fetch_token(self, url, **kwargs):68        resp = await self.post(url, **kwargs)69        text = await resp.aread()70        token = self.parse_response_token(resp.status_code, to_unicode(text))71        self.token = token72        return token73 74    @staticmethod75    def handle_error(error_type, error_description):76        raise OAuthError(error_type, error_description)77 78 79class OAuth1Client(_OAuth1Client, httpx.Client):80    auth_class = OAuth1Auth81 82    def __init__(self, client_id, client_secret=None,83                 token=None, token_secret=None,84                 redirect_uri=None, rsa_key=None, verifier=None,85                 signature_method=SIGNATURE_HMAC_SHA1,86                 signature_type=SIGNATURE_TYPE_HEADER,87                 force_include_body=False, **kwargs):88 89        _client_kwargs = extract_client_kwargs(kwargs)90        httpx.Client.__init__(self, **_client_kwargs)91 92        _OAuth1Client.__init__(93            self, self,94            client_id=client_id, client_secret=client_secret,95            token=token, token_secret=token_secret,96            redirect_uri=redirect_uri, rsa_key=rsa_key, verifier=verifier,97            signature_method=signature_method, signature_type=signature_type,98            force_include_body=force_include_body, **kwargs)99 100    @staticmethod101    def handle_error(error_type, error_description):102        raise OAuthError(error_type, error_description)103