codekingpro/portable-devtools

1import httpx2from httpx import Response, USE_CLIENT_DEFAULT3from authlib.oauth2.rfc7521 import AssertionClient as _AssertionClient4from authlib.oauth2.rfc7523 import JWTBearerGrant5from .utils import extract_client_kwargs6from .oauth2_client import OAuth2Auth7from ..base_client import OAuthError8 9__all__ = ['AsyncAssertionClient']10 11 12class AsyncAssertionClient(_AssertionClient, httpx.AsyncClient):13    token_auth_class = OAuth2Auth14    oauth_error_class = OAuthError15    JWT_BEARER_GRANT_TYPE = JWTBearerGrant.GRANT_TYPE16    ASSERTION_METHODS = {17        JWT_BEARER_GRANT_TYPE: JWTBearerGrant.sign,18    }19    DEFAULT_GRANT_TYPE = JWT_BEARER_GRANT_TYPE20 21    def __init__(self, token_endpoint, issuer, subject, audience=None, grant_type=None,22                 claims=None, token_placement='header', scope=None, **kwargs):23 24        client_kwargs = extract_client_kwargs(kwargs)25        httpx.AsyncClient.__init__(self, **client_kwargs)26 27        _AssertionClient.__init__(28            self, session=None,29            token_endpoint=token_endpoint, issuer=issuer, subject=subject,30            audience=audience, grant_type=grant_type, claims=claims,31            token_placement=token_placement, scope=scope, **kwargs32        )33 34    async def request(self, method, url, withhold_token=False, auth=USE_CLIENT_DEFAULT, **kwargs) -> Response:35        """Send request with auto refresh token feature."""36        if not withhold_token and auth is USE_CLIENT_DEFAULT:37            if not self.token or self.token.is_expired():38                await self.refresh_token()39 40            auth = self.token_auth41        return await super().request(42            method, url, auth=auth, **kwargs)43 44    async def _refresh_token(self, data):45        resp = await self.request(46            'POST', self.token_endpoint, data=data, withhold_token=True)47 48        return self.parse_response_token(resp)49 50 51class AssertionClient(_AssertionClient, httpx.Client):52    token_auth_class = OAuth2Auth53    oauth_error_class = OAuthError54    JWT_BEARER_GRANT_TYPE = JWTBearerGrant.GRANT_TYPE55    ASSERTION_METHODS = {56        JWT_BEARER_GRANT_TYPE: JWTBearerGrant.sign,57    }58    DEFAULT_GRANT_TYPE = JWT_BEARER_GRANT_TYPE59 60    def __init__(self, token_endpoint, issuer, subject, audience=None, grant_type=None,61                 claims=None, token_placement='header', scope=None, **kwargs):62 63        client_kwargs = extract_client_kwargs(kwargs)64        httpx.Client.__init__(self, **client_kwargs)65 66        _AssertionClient.__init__(67            self, session=self,68            token_endpoint=token_endpoint, issuer=issuer, subject=subject,69            audience=audience, grant_type=grant_type, claims=claims,70            token_placement=token_placement, scope=scope, **kwargs71        )72 73    def request(self, method, url, withhold_token=False, auth=USE_CLIENT_DEFAULT, **kwargs):74        """Send request with auto refresh token feature."""75        if not withhold_token and auth is USE_CLIENT_DEFAULT:76            if not self.token or self.token.is_expired():77                self.refresh_token()78 79            auth = self.token_auth80        return super().request(81            method, url, auth=auth, **kwargs)82