singer_sdk.authenticators.OAuthAuthenticator

class singer_sdk.authenticators.OAuthAuthenticator(stream: singer_sdk.streams.core.Stream, auth_endpoint: Optional[str] = None, oauth_scopes: Optional[str] = None, default_expiration: Optional[int] = None)

API Authenticator for OAuth 2.0 flows.

property auth_endpoint: str

Get the authorization endpoint.

Returns

The API authorization endpoint if it is set.

Raises

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.

Returns

HTTP headers for authentication.

property client_id: Optional[str]

Get client ID string to be used in authentication.

Returns

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

property client_secret: Optional[str]

Get client secret to be used in authentication.

Returns

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

is_token_valid() → bool

Check if token is valid.

Returns

True if the token is valid (fresh).

property oauth_request_body: dict

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

Get request body.

Returns

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

property oauth_scopes: Optional[str]

Get OAuth scopes.

Returns

String of OAuth scopes, or None if not set.

update_access_token() → None

Update access_token along with: last_refreshed and expires_in.

Raises

RuntimeError – When OAuth login fails.