singer_sdk.authenticators.OAuthAuthenticator#
- 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.
- Parameters:
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.
- update_access_token() None #
Update access_token along with: last_refreshed and expires_in.
- Raises:
RuntimeError – When OAuth login fails.
- 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: str | None#
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#
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#
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.