singer_sdk.authenticators.OAuthAuthenticator¶

class singer_sdk.authenticators.OAuthAuthenticator[source]¶

API Authenticator for OAuth 2.0 flows.

__init__(stream, auth_endpoint=None, oauth_scopes=None, default_expiration=None, oauth_headers=None)[source]¶

Create a new authenticator.

Parameters:

stream (RESTStream) – The stream instance to use with this authenticator.
auth_endpoint (str | None) – The OAuth 2.0 authorization endpoint.
oauth_scopes (str | None) – A comma-separated list of OAuth scopes.
default_expiration (int | None) – Default token expiry in seconds.
oauth_headers (dict | None) – An optional dict of headers required to get a token.

Return type:

None

authenticate_request(request)[source]¶

Authenticate an OAuth request.

Parameters:: request (PreparedRequest) – A requests.PreparedRequest object.
Returns:: The authenticated request object.
Return type:: PreparedRequest

is_token_valid()[source]¶

Check if token is valid.

Returns:: True if the token is valid (fresh).
Return type:: bool

update_access_token()[source]¶

Update access_token along with: last_refreshed and expires_in.

Raises:: RuntimeError – When OAuth login fails.
Return type:: None

property auth_endpoint: str[source]¶

Get the authorization endpoint.

Returns:: The API authorization endpoint if it is set.
Raises:: ValueError – If the endpoint is not set.

property client_id: str | None[source]¶

Get client ID string to be used in authentication.

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

property client_secret: str | None[source]¶

Get client secret to be used in authentication.

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

property oauth_request_body: dict[source]¶

Get formatted body of the OAuth authorization request.

Sample implementation:

@property
def oauth_request_body(self) -> dict:
    return {
        "grant_type": "password",
        "scope": "https://api.powerbi.com",
        "resource": "https://analysis.windows.net/powerbi/api",
        "client_id": self.config["client_id"],
        "username": self.config.get("username", self.config["client_id"]),
        "password": self.config["password"],
    }

Raises:: NotImplementedError – If derived class does not override this method.

property oauth_request_payload: dict[source]¶

Get request body.

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

property oauth_scopes: str | None[source]¶

Get OAuth scopes.

Returns:: String of OAuth scopes, or None if not set.