class singer_sdk.authenticators.OAuthAuthenticator#

API Authenticator for OAuth 2.0 flows.

__init__(stream: RESTStream, auth_endpoint: str | None = None, oauth_scopes: str | None = None, default_expiration: int | None = None, oauth_headers: dict | None = None) None#

Create a new authenticator.

  • stream – The stream instance to use with this authenticator.

  • auth_endpoint – The OAuth 2.0 authorization endpoint.

  • oauth_scopes – A comma-separated list of OAuth scopes.

  • default_expiration – Default token expiry in seconds.

  • oauth_headers – An optional dict of headers required to get a token.

property auth_endpoint: str#

Get the authorization endpoint.


The API authorization endpoint if it is set.


ValueError – If the endpoint is not set.

property auth_headers: dict#

Return a dictionary of auth headers to be applied.

These will be merged with any http_headers specified in the stream.


HTTP headers for authentication.

property client_id: str | None#

Get client ID string to be used in authentication.


Optional client secret from stream config if it has been set.

property client_secret: str | None#

Get client secret to be used in authentication.


Optional client secret from stream config if it has been set.

is_token_valid() bool#

Check if token is valid.


True if the token is valid (fresh).

property oauth_request_body: dict#

Get formatted body of the OAuth authorization request.

Sample implementation:

def oauth_request_body(self) -> dict:
    return {
        'grant_type': 'password',
        'scope': '',
        'resource': '',
        'client_id': self.config["client_id"],
        'username': self.config.get("username", self.config["client_id"]),
        'password': self.config["password"],

NotImplementedError – If derived class does not override this method.

property oauth_request_payload: dict#

Get request body.


A plain (OAuth) or encrypted (JWT) request body.

property oauth_scopes: str | None#

Get OAuth scopes.


String of OAuth scopes, or None if not set.

update_access_token() None#

Update access_token along with: last_refreshed and expires_in.


RuntimeError – When OAuth login fails.