Get Token

[Vad1mo]

In order to call an API with the bearer token, I need to get an /oauth/token first.
What is your suggestion on how to accomplish this with api-client?

specifically in context that the /oauth/token might expire and need to be renewed once the API server returns a 401.

[mom1]

Maybe Retrying help you.

[MikeWooster]

I don't think the client is yet capable of doing this. But not a bad idea for a new version. But in the meantime, I'd recommend maybe looking into retrying, but I'm not sure if that will have enough knowledge of the client implementation itself. Potentially you could extend the HeaderAuthentication so that it works out if the token is invalid or has expired and requests a new one? If you couple that with retrying on 401s, you might have a working solution.

[mom1]

This is most likely not the best solution, but still I'll leave it here:

# endpoints.py
from apiclient import endpoint


@endpoint(base_url='https://base.com')
class Endpoints:
    oauth_token = '/oauth/token'
    some1 = 'some1'
    some2 = 'some2'


# authentication_methods.py
from .endpoints import Endpoints


class HeaderAuthenticationRetry(HeaderAuthentication):
    def __init__(self, token: Optional[str] = None, **kwargs):
        super().__init__(token or '', **kwargs)

    def perform_initial_auth(self, client: APIClient, force: bool= Fasle):
        if force:
            self._token = client.get_token()
            # or
            self._token = client.get(Endpoints.oauth_token)

    def get_headers(self) -> Dict[str, str]:
        if self._token:
            return super().get_headers()
        return {}

# retrying.py
from apiclient.retrying import retry_if_api_request_error
from tenacity import RetryCallState, retry, stop_after_attempt, wait_fixed

from .authentication_methods import HeaderAuthenticationRetry


class retry_auth(retry_if_api_request_error):  # noqa: N801
    def __call__(self, retry_state: RetryCallState) -> bool:
        ret = super().__call__(retry_state=retry_state)
        if ret:
            client, *_ = retry_state.args
            auth = client.get_authentication_method()
            if isinstance(auth, HeaderAuthenticationRetry):
                auth._token = None
                auth.perform_initial_auth(client, force=True)
        return ret


use_token = retry(retry=retry_auth(status_codes=[401]), wait=wait_fixed(1), stop=stop_after_attempt(2), reraise=True)


# client.py
from .retrying import use_token
from .endpoints import Endpoints


class MyAPIClient(APIClient):
    def __init__(
        self,
        authentication_method: Optional[BaseAuthenticationMethod] = None,
        **kwargs,
    ):
        super().__init__(
            authentication_method=authentication_method or HeaderAuthenticationRetry(),
            **kwargs,
        )

    # Not auth
    def get_some1(self):
        return self.get(Endpoints.some1)

    # Auth
    @use_token
    def get_some2(self):
        return self.get(Endpoints.some2)

#  Use
client = MyAPIClient()
client.get_some1()  #  1 request without token
client.get_some2()  #  3 requests 1 - without token (401), 1 - get token, 1 - with token in header
client.get_some1()  #  1 request with token. Because it was saved in the previous request.