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.