airbyte_cdk.sources.streams.http.requests_native_auth 

 1#
 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 3#
 4
 5from .oauth import Oauth2Authenticator, SingleUseRefreshTokenOauth2Authenticator
 6from .protocols import ResponseAwareAuthenticator, TokenRotatingAuthenticator
 7from .token import BasicHttpAuthenticator, MultipleTokenAuthenticator, TokenAuthenticator
 8
 9__all__ = [
10    "Oauth2Authenticator",
11    "SingleUseRefreshTokenOauth2Authenticator",
12    "TokenAuthenticator",
13    "MultipleTokenAuthenticator",
14    "BasicHttpAuthenticator",
15    "ResponseAwareAuthenticator",
16    "TokenRotatingAuthenticator",
17]
class Oauth2Authenticator(airbyte_cdk.sources.streams.http.requests_native_auth.abstract_oauth.AbstractOauth2Authenticator): 
 27class Oauth2Authenticator(AbstractOauth2Authenticator):
 28    """
 29    Generates OAuth2.0 access tokens from an OAuth2.0 refresh token and client credentials.
 30    The generated access token is attached to each request via the Authorization header.
 31    If a connector_config is provided any mutation of it's value in the scope of this class will emit AirbyteControlConnectorConfigMessage.
 32    `refresh_token_error_messages` provides fixed user-facing messages for configured refresh token errors.
 33    """
 34
 35    def __init__(
 36        self,
 37        token_refresh_endpoint: str,
 38        client_id: str,
 39        client_secret: str,
 40        refresh_token: str,
 41        client_id_name: str = "client_id",
 42        client_secret_name: str = "client_secret",
 43        refresh_token_name: str = "refresh_token",
 44        scopes: List[str] | None = None,
 45        token_expiry_date: AirbyteDateTime | None = None,
 46        token_expiry_date_format: str | None = None,
 47        access_token_name: str = "access_token",
 48        expires_in_name: str = "expires_in",
 49        refresh_request_body: Mapping[str, Any] | None = None,
 50        refresh_request_headers: Mapping[str, Any] | None = None,
 51        send_refresh_request_as_query_params: bool = False,
 52        grant_type_name: str = "grant_type",
 53        grant_type: str = "refresh_token",
 54        token_expiry_is_time_of_expiration: bool = False,
 55        refresh_token_error_status_codes: Tuple[int, ...] = (),
 56        refresh_token_error_key: str = "",
 57        refresh_token_error_values: Tuple[str, ...] = (),
 58        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
 59    ) -> None:
 60        self._token_refresh_endpoint = token_refresh_endpoint
 61        self._client_secret_name = client_secret_name
 62        self._client_secret = client_secret
 63        self._client_id_name = client_id_name
 64        self._client_id = client_id
 65        self._refresh_token_name = refresh_token_name
 66        self._refresh_token = refresh_token
 67        self._scopes = scopes
 68        self._access_token_name = access_token_name
 69        self._expires_in_name = expires_in_name
 70        self._refresh_request_body = refresh_request_body
 71        self._refresh_request_headers = refresh_request_headers
 72        self._send_refresh_request_as_query_params = send_refresh_request_as_query_params
 73        self._grant_type_name = grant_type_name
 74        self._grant_type = grant_type
 75
 76        self._token_expiry_date = token_expiry_date or (ab_datetime_now() - timedelta(days=1))
 77        self._token_expiry_date_format = token_expiry_date_format
 78        self._token_expiry_is_time_of_expiration = token_expiry_is_time_of_expiration
 79        self._access_token = None
 80        super().__init__(
 81            refresh_token_error_status_codes,
 82            refresh_token_error_key,
 83            refresh_token_error_values,
 84            refresh_token_error_messages=refresh_token_error_messages,
 85        )
 86
 87    def get_token_refresh_endpoint(self) -> str:
 88        return self._token_refresh_endpoint
 89
 90    def get_client_id_name(self) -> str:
 91        return self._client_id_name
 92
 93    def get_client_id(self) -> str:
 94        return self._client_id
 95
 96    def get_client_secret_name(self) -> str:
 97        return self._client_secret_name
 98
 99    def get_client_secret(self) -> str:
100        return self._client_secret
101
102    def get_refresh_token_name(self) -> str:
103        return self._refresh_token_name
104
105    def get_refresh_token(self) -> str:
106        return self._refresh_token
107
108    def get_access_token_name(self) -> str:
109        return self._access_token_name
110
111    def get_scopes(self) -> list[str]:
112        return self._scopes  # type: ignore[return-value]
113
114    def get_expires_in_name(self) -> str:
115        return self._expires_in_name
116
117    def get_refresh_request_body(self) -> Mapping[str, Any]:
118        return self._refresh_request_body  # type: ignore[return-value]
119
120    def get_refresh_request_headers(self) -> Mapping[str, Any]:
121        return self._refresh_request_headers  # type: ignore[return-value]
122
123    def should_send_refresh_request_as_query_params(self) -> bool:
124        return self._send_refresh_request_as_query_params
125
126    def get_grant_type_name(self) -> str:
127        return self._grant_type_name
128
129    def get_grant_type(self) -> str:
130        return self._grant_type
131
132    def get_token_expiry_date(self) -> AirbyteDateTime:
133        return self._token_expiry_date
134
135    def set_token_expiry_date(self, value: AirbyteDateTime) -> None:
136        # Overrides of `refresh_access_token` written before 6.45.5 return the raw `expires_in`.
137        if not isinstance(value, datetime):
138            with suppress(ValueError, OverflowError):
139                value = self._parse_token_expiration_date(value)
140        self._token_expiry_date = value
141
142    @property
143    def token_expiry_is_time_of_expiration(self) -> bool:
144        return self._token_expiry_is_time_of_expiration
145
146    @property
147    def token_expiry_date_format(self) -> Optional[str]:
148        return self._token_expiry_date_format
149
150    @property
151    def access_token(self) -> str:
152        return self._access_token  # type: ignore[return-value]
153
154    @access_token.setter
155    def access_token(self, value: str) -> None:
156        self._access_token = value  # type: ignore[assignment]  # Incorrect type for assignment

Generates OAuth2.0 access tokens from an OAuth2.0 refresh token and client credentials. The generated access token is attached to each request via the Authorization header. If a connector_config is provided any mutation of it's value in the scope of this class will emit AirbyteControlConnectorConfigMessage. refresh_token_error_messages provides fixed user-facing messages for configured refresh token errors.

Oauth2Authenticator( token_refresh_endpoint: str, client_id: str, client_secret: str, refresh_token: str, client_id_name: str = 'client_id', client_secret_name: str = 'client_secret', refresh_token_name: str = 'refresh_token', scopes: Optional[List[str]] = None, token_expiry_date: airbyte_cdk.utils.datetime_helpers.AirbyteDateTime | None = None, token_expiry_date_format: str | None = None, access_token_name: str = 'access_token', expires_in_name: str = 'expires_in', refresh_request_body: Optional[Mapping[str, Any]] = None, refresh_request_headers: Optional[Mapping[str, Any]] = None, send_refresh_request_as_query_params: bool = False, grant_type_name: str = 'grant_type', grant_type: str = 'refresh_token', token_expiry_is_time_of_expiration: bool = False, refresh_token_error_status_codes: Tuple[int, ...] = (), refresh_token_error_key: str = '', refresh_token_error_values: Tuple[str, ...] = (), refresh_token_error_messages: Optional[Mapping[str, str]] = None) 
35    def __init__(
36        self,
37        token_refresh_endpoint: str,
38        client_id: str,
39        client_secret: str,
40        refresh_token: str,
41        client_id_name: str = "client_id",
42        client_secret_name: str = "client_secret",
43        refresh_token_name: str = "refresh_token",
44        scopes: List[str] | None = None,
45        token_expiry_date: AirbyteDateTime | None = None,
46        token_expiry_date_format: str | None = None,
47        access_token_name: str = "access_token",
48        expires_in_name: str = "expires_in",
49        refresh_request_body: Mapping[str, Any] | None = None,
50        refresh_request_headers: Mapping[str, Any] | None = None,
51        send_refresh_request_as_query_params: bool = False,
52        grant_type_name: str = "grant_type",
53        grant_type: str = "refresh_token",
54        token_expiry_is_time_of_expiration: bool = False,
55        refresh_token_error_status_codes: Tuple[int, ...] = (),
56        refresh_token_error_key: str = "",
57        refresh_token_error_values: Tuple[str, ...] = (),
58        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
59    ) -> None:
60        self._token_refresh_endpoint = token_refresh_endpoint
61        self._client_secret_name = client_secret_name
62        self._client_secret = client_secret
63        self._client_id_name = client_id_name
64        self._client_id = client_id
65        self._refresh_token_name = refresh_token_name
66        self._refresh_token = refresh_token
67        self._scopes = scopes
68        self._access_token_name = access_token_name
69        self._expires_in_name = expires_in_name
70        self._refresh_request_body = refresh_request_body
71        self._refresh_request_headers = refresh_request_headers
72        self._send_refresh_request_as_query_params = send_refresh_request_as_query_params
73        self._grant_type_name = grant_type_name
74        self._grant_type = grant_type
75
76        self._token_expiry_date = token_expiry_date or (ab_datetime_now() - timedelta(days=1))
77        self._token_expiry_date_format = token_expiry_date_format
78        self._token_expiry_is_time_of_expiration = token_expiry_is_time_of_expiration
79        self._access_token = None
80        super().__init__(
81            refresh_token_error_status_codes,
82            refresh_token_error_key,
83            refresh_token_error_values,
84            refresh_token_error_messages=refresh_token_error_messages,
85        )

If all of refresh_token_error_status_codes, refresh_token_error_key, and refresh_token_error_values are set, then http errors with such params will be wrapped in AirbyteTracedException. refresh_token_error_messages maps fixed provider error strings to fixed user-facing messages.

def get_token_refresh_endpoint(self) -> str: 
87    def get_token_refresh_endpoint(self) -> str:
88        return self._token_refresh_endpoint

Returns the endpoint to refresh the access token

def get_client_id_name(self) -> str: 
90    def get_client_id_name(self) -> str:
91        return self._client_id_name

The client id name to authenticate

def get_client_id(self) -> str: 
93    def get_client_id(self) -> str:
94        return self._client_id

The client id to authenticate

def get_client_secret_name(self) -> str: 
96    def get_client_secret_name(self) -> str:
97        return self._client_secret_name

The client secret name to authenticate

def get_client_secret(self) -> str: 
 99    def get_client_secret(self) -> str:
100        return self._client_secret

The client secret to authenticate

def get_refresh_token_name(self) -> str: 
102    def get_refresh_token_name(self) -> str:
103        return self._refresh_token_name

The refresh token name to authenticate

def get_refresh_token(self) -> str: 
105    def get_refresh_token(self) -> str:
106        return self._refresh_token

The token used to refresh the access token when it expires

def get_access_token_name(self) -> str: 
108    def get_access_token_name(self) -> str:
109        return self._access_token_name

Field to extract access token from in the response

def get_scopes(self) -> list[str]: 
111    def get_scopes(self) -> list[str]:
112        return self._scopes  # type: ignore[return-value]

List of requested scopes

def get_expires_in_name(self) -> str: 
114    def get_expires_in_name(self) -> str:
115        return self._expires_in_name

Returns the expires_in field name

def get_refresh_request_body(self) -> Mapping[str, Any]: 
117    def get_refresh_request_body(self) -> Mapping[str, Any]:
118        return self._refresh_request_body  # type: ignore[return-value]

Returns the request body to set on the refresh request

def get_refresh_request_headers(self) -> Mapping[str, Any]: 
120    def get_refresh_request_headers(self) -> Mapping[str, Any]:
121        return self._refresh_request_headers  # type: ignore[return-value]

Returns the request headers to set on the refresh request

def should_send_refresh_request_as_query_params(self) -> bool: 
123    def should_send_refresh_request_as_query_params(self) -> bool:
124        return self._send_refresh_request_as_query_params

Returns True if the standard refresh args should be sent on the URL query string instead of in the request body.

Defaults to False so existing authenticators retain their previous behavior (params in body, no query params on the refresh URL). Subclasses can override this to opt into the URL-query-string shape required by OAuth providers like Gong.

def get_grant_type_name(self) -> str: 
126    def get_grant_type_name(self) -> str:
127        return self._grant_type_name

Returns grant_type specified name for requesting access_token

def get_grant_type(self) -> str: 
129    def get_grant_type(self) -> str:
130        return self._grant_type

Returns grant_type specified for requesting access_token

def get_token_expiry_date(self) -> airbyte_cdk.utils.datetime_helpers.AirbyteDateTime: 
132    def get_token_expiry_date(self) -> AirbyteDateTime:
133        return self._token_expiry_date

Expiration date of the access token

def set_token_expiry_date(self, value: airbyte_cdk.utils.datetime_helpers.AirbyteDateTime) -> None: 
135    def set_token_expiry_date(self, value: AirbyteDateTime) -> None:
136        # Overrides of `refresh_access_token` written before 6.45.5 return the raw `expires_in`.
137        if not isinstance(value, datetime):
138            with suppress(ValueError, OverflowError):
139                value = self._parse_token_expiration_date(value)
140        self._token_expiry_date = value

Setter for access token expiration date

token_expiry_is_time_of_expiration: bool 
142    @property
143    def token_expiry_is_time_of_expiration(self) -> bool:
144        return self._token_expiry_is_time_of_expiration

Indicates that the Token Expiry returns the date until which the token will be valid, not the amount of time it will be valid.

token_expiry_date_format: Optional[str] 
146    @property
147    def token_expiry_date_format(self) -> Optional[str]:
148        return self._token_expiry_date_format

Format of the datetime; exists it if expires_in is returned as the expiration datetime instead of seconds until it expires

access_token: str 
150    @property
151    def access_token(self) -> str:
152        return self._access_token  # type: ignore[return-value]

Returns the access token

Inherited Members
airbyte_cdk.sources.streams.http.requests_native_auth.abstract_oauth.AbstractOauth2Authenticator
get_auth_header
get_access_token
refresh_and_set_access_token
token_has_expired
build_refresh_request_body
build_refresh_request_headers
build_refresh_request_query_params
refresh_access_token
class SingleUseRefreshTokenOauth2Authenticator(airbyte_cdk.sources.streams.http.requests_native_auth.Oauth2Authenticator): 
159class SingleUseRefreshTokenOauth2Authenticator(Oauth2Authenticator):
160    """
161    Authenticator that should be used for API implementing single use refresh tokens:
162    when refreshing access token some API returns a new refresh token that needs to used in the next refresh flow.
163    This authenticator updates the configuration with new refresh token by emitting Airbyte control message from an observed mutation.
164    By default, this authenticator expects a connector config with a "credentials" field with the following nested fields: client_id,
165    client_secret, refresh_token. This behavior can be changed by defining custom config path (using dpath paths) in client_id_config_path,
166    client_secret_config_path, refresh_token_config_path constructor arguments.
167    `refresh_token_error_messages` provides fixed user-facing messages for configured refresh token errors.
168    """
169
170    def __init__(
171        self,
172        connector_config: Mapping[str, Any],
173        token_refresh_endpoint: str,
174        scopes: List[str] | None = None,
175        access_token_name: str = "access_token",
176        expires_in_name: str = "expires_in",
177        refresh_token_name: str = "refresh_token",
178        refresh_request_body: Mapping[str, Any] | None = None,
179        refresh_request_headers: Mapping[str, Any] | None = None,
180        send_refresh_request_as_query_params: bool = False,
181        grant_type_name: str = "grant_type",
182        grant_type: str = "refresh_token",
183        client_id_name: str = "client_id",
184        client_id: Optional[str] = None,
185        client_secret_name: str = "client_secret",
186        client_secret: Optional[str] = None,
187        access_token_config_path: Sequence[str] = ("credentials", "access_token"),
188        refresh_token_config_path: Sequence[str] = ("credentials", "refresh_token"),
189        token_expiry_date_config_path: Sequence[str] = ("credentials", "token_expiry_date"),
190        token_expiry_date_format: Optional[str] = None,
191        message_repository: MessageRepository = NoopMessageRepository(),
192        token_expiry_is_time_of_expiration: bool = False,
193        refresh_token_error_status_codes: Tuple[int, ...] = (),
194        refresh_token_error_key: str = "",
195        refresh_token_error_values: Tuple[str, ...] = (),
196        emit_control_message_to_message_repository: bool = False,
197        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
198    ) -> None:
199        """
200        Args:
201            connector_config (Mapping[str, Any]): The full connector configuration
202            token_refresh_endpoint (str): Full URL to the token refresh endpoint
203            scopes (List[str], optional): List of OAuth scopes to pass in the refresh token request body. Defaults to None.
204            access_token_name (str, optional): Name of the access token field, used to parse the refresh token response. Defaults to "access_token".
205            expires_in_name (str, optional): Name of the name of the field that characterizes when the current access token will expire, used to parse the refresh token response. Defaults to "expires_in".
206            refresh_token_name (str, optional): Name of the name of the refresh token field, used to parse the refresh token response. Defaults to "refresh_token".
207            refresh_request_body (Mapping[str, Any], optional): Custom key value pair that will be added to the refresh token request body. Defaults to None.
208            refresh_request_headers (Mapping[str, Any], optional): Custom key value pair that will be added to the refresh token request headers. Defaults to None.
209            send_refresh_request_as_query_params (bool, optional): When True, the standard refresh args (`grant_type`, `refresh_token`, client credentials when not in an `Authorization` header, scopes, plus any `refresh_request_body` extras) are sent on the URL query string and the request body is emitted empty. Use this for OAuth providers like Gong that document their refresh endpoint with refresh args on the URL query string. Defaults to False.
210            grant_type (str, optional): OAuth grant type. Defaults to "refresh_token".
211            client_id (Optional[str]): The client id to authenticate. If not specified, defaults to credentials.client_id in the config object.
212            client_secret (Optional[str]): The client secret to authenticate. If not specified, defaults to credentials.client_secret in the config object.
213            access_token_config_path (Sequence[str]): Dpath to the access_token field in the connector configuration. Defaults to ("credentials", "access_token").
214            refresh_token_config_path (Sequence[str]): Dpath to the refresh_token field in the connector configuration. Defaults to ("credentials", "refresh_token").
215            token_expiry_date_config_path (Sequence[str]): Dpath to the token_expiry_date field in the connector configuration. Defaults to ("credentials", "token_expiry_date").
216            token_expiry_date_format (Optional[str]): Date format of the token expiry date field (set by expires_in_name). If specified the token expiry date is interpreted as time of expiration, as with token_expiry_is_time_of_expiration, otherwise as number of seconds until expiration.
217            token_expiry_is_time_of_expiration bool: set True it if expires_in is returned as time of expiration instead of the number seconds until expiration
218            message_repository (MessageRepository): The message repository used to emit logs on HTTP requests and control message on config update.
219            emit_control_message_to_message_repository (bool): Whether the message repository additionally receives the CONNECTOR_CONFIG message. This is only for in-process consumers such as the Connector Builder, because messages in the repository are otherwise printed to stdout by the entrypoint and would duplicate the direct stdout emission. Defaults to False.
220            refresh_token_error_messages (Optional[Mapping[str, str]]): Fixed user-facing messages for refresh token errors
221        """
222        self._connector_config = connector_config
223        self._client_id: str = self._get_config_value_by_path(
224            ("credentials", "client_id"), client_id
225        )
226        self._client_secret: str = self._get_config_value_by_path(
227            ("credentials", "client_secret"), client_secret
228        )
229        self._client_id_name = client_id_name
230        self._client_secret_name = client_secret_name
231        self._access_token_config_path = access_token_config_path
232        self._refresh_token_config_path = refresh_token_config_path
233        self._token_expiry_date_config_path = token_expiry_date_config_path
234        self._token_expiry_date_format = token_expiry_date_format
235        self._refresh_token_name = refresh_token_name
236        self._grant_type_name = grant_type_name
237        self._connector_config = connector_config
238        self.__message_repository = message_repository
239        self._emit_control_message_to_message_repository = (
240            emit_control_message_to_message_repository
241        )
242        super().__init__(
243            token_refresh_endpoint=token_refresh_endpoint,
244            client_id_name=self._client_id_name,
245            client_id=self._client_id,
246            client_secret_name=self._client_secret_name,
247            client_secret=self._client_secret,
248            refresh_token=self.get_refresh_token(),
249            refresh_token_name=self._refresh_token_name,
250            scopes=scopes,
251            token_expiry_date=self.get_token_expiry_date(),
252            access_token_name=access_token_name,
253            expires_in_name=expires_in_name,
254            refresh_request_body=refresh_request_body,
255            refresh_request_headers=refresh_request_headers,
256            send_refresh_request_as_query_params=send_refresh_request_as_query_params,
257            grant_type_name=self._grant_type_name,
258            grant_type=grant_type,
259            token_expiry_date_format=token_expiry_date_format,
260            token_expiry_is_time_of_expiration=token_expiry_is_time_of_expiration
261            or bool(token_expiry_date_format),
262            refresh_token_error_status_codes=refresh_token_error_status_codes,
263            refresh_token_error_key=refresh_token_error_key,
264            refresh_token_error_values=refresh_token_error_values,
265            refresh_token_error_messages=refresh_token_error_messages,
266        )
267
268    @property
269    def access_token(self) -> str:
270        """
271        Retrieve the access token from the configuration.
272
273        Returns:
274            str: The access token.
275        """
276        return self._get_config_value_by_path(self._access_token_config_path)  # type: ignore[return-value]
277
278    @access_token.setter
279    def access_token(self, new_access_token: str) -> None:
280        """
281        Sets a new access token.
282
283        Args:
284            new_access_token (str): The new access token to be set.
285        """
286        self._set_config_value_by_path(self._access_token_config_path, new_access_token)
287
288    def get_refresh_token(self) -> str:
289        """
290        Retrieve the refresh token from the configuration.
291
292        This method fetches the refresh token using the configuration path specified
293        by `_refresh_token_config_path`.
294
295        Returns:
296            str: The refresh token as a string.
297        """
298        return self._get_config_value_by_path(self._refresh_token_config_path)  # type: ignore[return-value]
299
300    def set_refresh_token(self, new_refresh_token: str) -> None:
301        """
302        Updates the refresh token in the configuration.
303
304        Args:
305            new_refresh_token (str): The new refresh token to be set.
306        """
307        self._set_config_value_by_path(self._refresh_token_config_path, new_refresh_token)
308
309    def get_token_expiry_date(self) -> AirbyteDateTime:
310        """
311        Retrieves the token expiry date from the configuration.
312
313        This method fetches the token expiry date from the configuration using the specified path.
314        If the expiry date is an empty string, it returns the current date and time minus one day.
315        Otherwise, it parses the expiry date string into an AirbyteDateTime object.
316
317        Returns:
318            AirbyteDateTime: The parsed or calculated token expiry date.
319
320        Raises:
321            TypeError: If the result is not an instance of AirbyteDateTime.
322        """
323        expiry_date = self._get_config_value_by_path(self._token_expiry_date_config_path)
324        result = (
325            ab_datetime_now() - timedelta(days=1)
326            if expiry_date == ""
327            else ab_datetime_parse(str(expiry_date))
328        )
329        if isinstance(result, AirbyteDateTime):
330            return result
331        raise TypeError("Invalid datetime conversion")
332
333    def set_token_expiry_date(self, new_token_expiry_date: AirbyteDateTime) -> None:  # type: ignore[override]
334        """
335        Sets the token expiry date in the configuration.
336
337        Args:
338            new_token_expiry_date (AirbyteDateTime): The new expiry date for the token.
339        """
340        if not (
341            isinstance(new_token_expiry_date, datetime) or self.token_expiry_is_time_of_expiration
342        ):
343            # Overrides of `refresh_access_token` written before 6.45.5 return the raw `expires_in`
344            # in seconds. Other values are stored as-is and parsed on read.
345            with suppress(ValueError, OverflowError):
346                new_token_expiry_date = self._parse_token_expiration_date(new_token_expiry_date)
347        self._set_config_value_by_path(
348            self._token_expiry_date_config_path, str(new_token_expiry_date)
349        )
350
351    def token_has_expired(self) -> bool:
352        """Returns True if the token is expired"""
353        return ab_datetime_now() > self.get_token_expiry_date()
354
355    def get_access_token(self) -> str:
356        """Retrieve new access and refresh token if the access token has expired.
357
358        This method uses double-checked locking to ensure thread-safe token refresh.
359        This is especially critical for single-use refresh tokens where concurrent
360        refresh attempts would cause failures as the refresh token is invalidated
361        after first use.
362
363        The new refresh token is persisted with the set_refresh_token function.
364
365        Returns:
366            str: The current access_token, updated if it was previously expired.
367        """
368        if self.token_has_expired():
369            with self._token_refresh_lock:
370                # Double-check after acquiring lock - another thread may have already refreshed
371                if self.token_has_expired():
372                    self.refresh_and_set_access_token()
373        return self.access_token
374
375    def refresh_and_set_access_token(self) -> None:
376        """Force refresh the access token and update internal state.
377
378        For single-use refresh tokens, this also persists the new refresh token
379        and emits a control message to update the connector config. If the
380        response omits a refresh token, the existing one is preserved.
381        """
382        token_before_waiting = self._current_access_token_or_none()
383        with self._token_refresh_lock:
384            if self._current_access_token_or_none() != token_before_waiting:
385                return
386            new_access_token, access_token_expires_in, new_refresh_token = (
387                self.refresh_access_token()
388            )
389            self.access_token = new_access_token
390            if new_refresh_token is not None:
391                self.set_refresh_token(new_refresh_token)
392            self.set_token_expiry_date(access_token_expires_in)
393            self._emit_control_message()
394
395    def refresh_access_token(self) -> Tuple[str, AirbyteDateTime, Optional[str]]:  # type: ignore[override]
396        """
397        Refreshes the access token by making a handled request and extracting the necessary token information.
398
399        Returns:
400            A tuple of (access_token, token_expiry_date, refresh_token). The refresh token
401            is `None` when the OAuth provider omits it from the response.
402        """
403        response_json = self._make_handled_request()
404        return (
405            self._extract_access_token(response_json),
406            self._extract_token_expiry_date(response_json),
407            self._extract_refresh_token(response_json),
408        )
409
410    def _set_config_value_by_path(self, config_path: Union[str, Sequence[str]], value: Any) -> None:
411        """
412        Set a value in the connector configuration at the specified path.
413
414        Args:
415            config_path (Union[str, Sequence[str]]): The path within the configuration where the value should be set.
416                This can be a string representing a single key or a sequence of strings representing a nested path.
417            value (Any): The value to set at the specified path in the configuration.
418
419        Returns:
420            None
421        """
422        dpath.new(self._connector_config, config_path, value)  # type: ignore[arg-type]
423
424    def _get_config_value_by_path(
425        self, config_path: Union[str, Sequence[str]], default: Optional[str] = None
426    ) -> str | Any:
427        """
428        Retrieve a value from the connector configuration using a specified path.
429
430        Args:
431            config_path (Union[str, Sequence[str]]): The path to the desired configuration value. This can be a string or a sequence of strings.
432            default (Optional[str], optional): The default value to return if the specified path does not exist in the configuration. Defaults to None.
433
434        Returns:
435            Any: The value from the configuration at the specified path, or the default value if the path does not exist.
436        """
437        return dpath.get(
438            self._connector_config,  # type: ignore[arg-type]
439            config_path,
440            default=default if default is not None else "",
441        )
442
443    def _emit_control_message(self) -> None:
444        """Emit the updated connector config as a CONNECTOR_CONFIG control message.
445
446        The message is always printed to stdout, which is the delivery the platform relies on: it
447        is immediate and independent of whether the source exposes a message repository and of
448        when that repository is drained (for `ConcurrentDeclarativeSource` the internal repository
449        is only drained during `read`, after partitions are processed). The message repository is
450        only used in addition when explicitly requested, for in-process consumers such as the
451        Connector Builder; repository messages are otherwise printed by the entrypoint too and
452        would duplicate the stdout emission.
453        """
454        emit_configuration_as_airbyte_control_message(self._connector_config)  # type: ignore[arg-type]
455        if self._emit_control_message_to_message_repository and not isinstance(
456            self._message_repository, NoopMessageRepository
457        ):
458            self._message_repository.emit_message(
459                create_connector_config_control_message(self._connector_config)  # type: ignore[arg-type]
460            )
461
462    @property
463    def _message_repository(self) -> MessageRepository:
464        """
465        Overriding AbstractOauth2Authenticator._message_repository to allow for HTTP request logs
466        """
467        return self.__message_repository

Authenticator that should be used for API implementing single use refresh tokens: when refreshing access token some API returns a new refresh token that needs to used in the next refresh flow. This authenticator updates the configuration with new refresh token by emitting Airbyte control message from an observed mutation. By default, this authenticator expects a connector config with a "credentials" field with the following nested fields: client_id, client_secret, refresh_token. This behavior can be changed by defining custom config path (using dpath paths) in client_id_config_path, client_secret_config_path, refresh_token_config_path constructor arguments. refresh_token_error_messages provides fixed user-facing messages for configured refresh token errors.

SingleUseRefreshTokenOauth2Authenticator( connector_config: Mapping[str, Any], token_refresh_endpoint: str, scopes: Optional[List[str]] = None, access_token_name: str = 'access_token', expires_in_name: str = 'expires_in', refresh_token_name: str = 'refresh_token', refresh_request_body: Optional[Mapping[str, Any]] = None, refresh_request_headers: Optional[Mapping[str, Any]] = None, send_refresh_request_as_query_params: bool = False, grant_type_name: str = 'grant_type', grant_type: str = 'refresh_token', client_id_name: str = 'client_id', client_id: Optional[str] = None, client_secret_name: str = 'client_secret', client_secret: Optional[str] = None, access_token_config_path: Sequence[str] = ('credentials', 'access_token'), refresh_token_config_path: Sequence[str] = ('credentials', 'refresh_token'), token_expiry_date_config_path: Sequence[str] = ('credentials', 'token_expiry_date'), token_expiry_date_format: Optional[str] = None, message_repository: airbyte_cdk.MessageRepository = <airbyte_cdk.sources.message.NoopMessageRepository object>, token_expiry_is_time_of_expiration: bool = False, refresh_token_error_status_codes: Tuple[int, ...] = (), refresh_token_error_key: str = '', refresh_token_error_values: Tuple[str, ...] = (), emit_control_message_to_message_repository: bool = False, refresh_token_error_messages: Optional[Mapping[str, str]] = None) 
170    def __init__(
171        self,
172        connector_config: Mapping[str, Any],
173        token_refresh_endpoint: str,
174        scopes: List[str] | None = None,
175        access_token_name: str = "access_token",
176        expires_in_name: str = "expires_in",
177        refresh_token_name: str = "refresh_token",
178        refresh_request_body: Mapping[str, Any] | None = None,
179        refresh_request_headers: Mapping[str, Any] | None = None,
180        send_refresh_request_as_query_params: bool = False,
181        grant_type_name: str = "grant_type",
182        grant_type: str = "refresh_token",
183        client_id_name: str = "client_id",
184        client_id: Optional[str] = None,
185        client_secret_name: str = "client_secret",
186        client_secret: Optional[str] = None,
187        access_token_config_path: Sequence[str] = ("credentials", "access_token"),
188        refresh_token_config_path: Sequence[str] = ("credentials", "refresh_token"),
189        token_expiry_date_config_path: Sequence[str] = ("credentials", "token_expiry_date"),
190        token_expiry_date_format: Optional[str] = None,
191        message_repository: MessageRepository = NoopMessageRepository(),
192        token_expiry_is_time_of_expiration: bool = False,
193        refresh_token_error_status_codes: Tuple[int, ...] = (),
194        refresh_token_error_key: str = "",
195        refresh_token_error_values: Tuple[str, ...] = (),
196        emit_control_message_to_message_repository: bool = False,
197        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
198    ) -> None:
199        """
200        Args:
201            connector_config (Mapping[str, Any]): The full connector configuration
202            token_refresh_endpoint (str): Full URL to the token refresh endpoint
203            scopes (List[str], optional): List of OAuth scopes to pass in the refresh token request body. Defaults to None.
204            access_token_name (str, optional): Name of the access token field, used to parse the refresh token response. Defaults to "access_token".
205            expires_in_name (str, optional): Name of the name of the field that characterizes when the current access token will expire, used to parse the refresh token response. Defaults to "expires_in".
206            refresh_token_name (str, optional): Name of the name of the refresh token field, used to parse the refresh token response. Defaults to "refresh_token".
207            refresh_request_body (Mapping[str, Any], optional): Custom key value pair that will be added to the refresh token request body. Defaults to None.
208            refresh_request_headers (Mapping[str, Any], optional): Custom key value pair that will be added to the refresh token request headers. Defaults to None.
209            send_refresh_request_as_query_params (bool, optional): When True, the standard refresh args (`grant_type`, `refresh_token`, client credentials when not in an `Authorization` header, scopes, plus any `refresh_request_body` extras) are sent on the URL query string and the request body is emitted empty. Use this for OAuth providers like Gong that document their refresh endpoint with refresh args on the URL query string. Defaults to False.
210            grant_type (str, optional): OAuth grant type. Defaults to "refresh_token".
211            client_id (Optional[str]): The client id to authenticate. If not specified, defaults to credentials.client_id in the config object.
212            client_secret (Optional[str]): The client secret to authenticate. If not specified, defaults to credentials.client_secret in the config object.
213            access_token_config_path (Sequence[str]): Dpath to the access_token field in the connector configuration. Defaults to ("credentials", "access_token").
214            refresh_token_config_path (Sequence[str]): Dpath to the refresh_token field in the connector configuration. Defaults to ("credentials", "refresh_token").
215            token_expiry_date_config_path (Sequence[str]): Dpath to the token_expiry_date field in the connector configuration. Defaults to ("credentials", "token_expiry_date").
216            token_expiry_date_format (Optional[str]): Date format of the token expiry date field (set by expires_in_name). If specified the token expiry date is interpreted as time of expiration, as with token_expiry_is_time_of_expiration, otherwise as number of seconds until expiration.
217            token_expiry_is_time_of_expiration bool: set True it if expires_in is returned as time of expiration instead of the number seconds until expiration
218            message_repository (MessageRepository): The message repository used to emit logs on HTTP requests and control message on config update.
219            emit_control_message_to_message_repository (bool): Whether the message repository additionally receives the CONNECTOR_CONFIG message. This is only for in-process consumers such as the Connector Builder, because messages in the repository are otherwise printed to stdout by the entrypoint and would duplicate the direct stdout emission. Defaults to False.
220            refresh_token_error_messages (Optional[Mapping[str, str]]): Fixed user-facing messages for refresh token errors
221        """
222        self._connector_config = connector_config
223        self._client_id: str = self._get_config_value_by_path(
224            ("credentials", "client_id"), client_id
225        )
226        self._client_secret: str = self._get_config_value_by_path(
227            ("credentials", "client_secret"), client_secret
228        )
229        self._client_id_name = client_id_name
230        self._client_secret_name = client_secret_name
231        self._access_token_config_path = access_token_config_path
232        self._refresh_token_config_path = refresh_token_config_path
233        self._token_expiry_date_config_path = token_expiry_date_config_path
234        self._token_expiry_date_format = token_expiry_date_format
235        self._refresh_token_name = refresh_token_name
236        self._grant_type_name = grant_type_name
237        self._connector_config = connector_config
238        self.__message_repository = message_repository
239        self._emit_control_message_to_message_repository = (
240            emit_control_message_to_message_repository
241        )
242        super().__init__(
243            token_refresh_endpoint=token_refresh_endpoint,
244            client_id_name=self._client_id_name,
245            client_id=self._client_id,
246            client_secret_name=self._client_secret_name,
247            client_secret=self._client_secret,
248            refresh_token=self.get_refresh_token(),
249            refresh_token_name=self._refresh_token_name,
250            scopes=scopes,
251            token_expiry_date=self.get_token_expiry_date(),
252            access_token_name=access_token_name,
253            expires_in_name=expires_in_name,
254            refresh_request_body=refresh_request_body,
255            refresh_request_headers=refresh_request_headers,
256            send_refresh_request_as_query_params=send_refresh_request_as_query_params,
257            grant_type_name=self._grant_type_name,
258            grant_type=grant_type,
259            token_expiry_date_format=token_expiry_date_format,
260            token_expiry_is_time_of_expiration=token_expiry_is_time_of_expiration
261            or bool(token_expiry_date_format),
262            refresh_token_error_status_codes=refresh_token_error_status_codes,
263            refresh_token_error_key=refresh_token_error_key,
264            refresh_token_error_values=refresh_token_error_values,
265            refresh_token_error_messages=refresh_token_error_messages,
266        )
Arguments:
  • connector_config (Mapping[str, Any]): The full connector configuration
  • token_refresh_endpoint (str): Full URL to the token refresh endpoint
  • scopes (List[str], optional): List of OAuth scopes to pass in the refresh token request body. Defaults to None.
  • access_token_name (str, optional): Name of the access token field, used to parse the refresh token response. Defaults to "access_token".
  • expires_in_name (str, optional): Name of the name of the field that characterizes when the current access token will expire, used to parse the refresh token response. Defaults to "expires_in".
  • refresh_token_name (str, optional): Name of the name of the refresh token field, used to parse the refresh token response. Defaults to "refresh_token".
  • refresh_request_body (Mapping[str, Any], optional): Custom key value pair that will be added to the refresh token request body. Defaults to None.
  • refresh_request_headers (Mapping[str, Any], optional): Custom key value pair that will be added to the refresh token request headers. Defaults to None.
  • send_refresh_request_as_query_params (bool, optional): When True, the standard refresh args (grant_type, refresh_token, client credentials when not in an Authorization header, scopes, plus any refresh_request_body extras) are sent on the URL query string and the request body is emitted empty. Use this for OAuth providers like Gong that document their refresh endpoint with refresh args on the URL query string. Defaults to False.
  • grant_type (str, optional): OAuth grant type. Defaults to "refresh_token".
  • client_id (Optional[str]): The client id to authenticate. If not specified, defaults to credentials.client_id in the config object.
  • client_secret (Optional[str]): The client secret to authenticate. If not specified, defaults to credentials.client_secret in the config object.
  • access_token_config_path (Sequence[str]): Dpath to the access_token field in the connector configuration. Defaults to ("credentials", "access_token").
  • refresh_token_config_path (Sequence[str]): Dpath to the refresh_token field in the connector configuration. Defaults to ("credentials", "refresh_token").
  • token_expiry_date_config_path (Sequence[str]): Dpath to the token_expiry_date field in the connector configuration. Defaults to ("credentials", "token_expiry_date").
  • token_expiry_date_format (Optional[str]): Date format of the token expiry date field (set by expires_in_name). If specified the token expiry date is interpreted as time of expiration, as with token_expiry_is_time_of_expiration, otherwise as number of seconds until expiration.
  • token_expiry_is_time_of_expiration bool: set True it if expires_in is returned as time of expiration instead of the number seconds until expiration
  • message_repository (MessageRepository): The message repository used to emit logs on HTTP requests and control message on config update.
  • emit_control_message_to_message_repository (bool): Whether the message repository additionally receives the CONNECTOR_CONFIG message. This is only for in-process consumers such as the Connector Builder, because messages in the repository are otherwise printed to stdout by the entrypoint and would duplicate the direct stdout emission. Defaults to False.
  • refresh_token_error_messages (Optional[Mapping[str, str]]): Fixed user-facing messages for refresh token errors
access_token: str 
268    @property
269    def access_token(self) -> str:
270        """
271        Retrieve the access token from the configuration.
272
273        Returns:
274            str: The access token.
275        """
276        return self._get_config_value_by_path(self._access_token_config_path)  # type: ignore[return-value]

Retrieve the access token from the configuration.

Returns:

str: The access token.

def get_refresh_token(self) -> str: 
288    def get_refresh_token(self) -> str:
289        """
290        Retrieve the refresh token from the configuration.
291
292        This method fetches the refresh token using the configuration path specified
293        by `_refresh_token_config_path`.
294
295        Returns:
296            str: The refresh token as a string.
297        """
298        return self._get_config_value_by_path(self._refresh_token_config_path)  # type: ignore[return-value]

Retrieve the refresh token from the configuration.

This method fetches the refresh token using the configuration path specified by _refresh_token_config_path.

Returns:

str: The refresh token as a string.

def set_refresh_token(self, new_refresh_token: str) -> None: 
300    def set_refresh_token(self, new_refresh_token: str) -> None:
301        """
302        Updates the refresh token in the configuration.
303
304        Args:
305            new_refresh_token (str): The new refresh token to be set.
306        """
307        self._set_config_value_by_path(self._refresh_token_config_path, new_refresh_token)

Updates the refresh token in the configuration.

Arguments:
  • new_refresh_token (str): The new refresh token to be set.
def get_token_expiry_date(self) -> airbyte_cdk.utils.datetime_helpers.AirbyteDateTime: 
309    def get_token_expiry_date(self) -> AirbyteDateTime:
310        """
311        Retrieves the token expiry date from the configuration.
312
313        This method fetches the token expiry date from the configuration using the specified path.
314        If the expiry date is an empty string, it returns the current date and time minus one day.
315        Otherwise, it parses the expiry date string into an AirbyteDateTime object.
316
317        Returns:
318            AirbyteDateTime: The parsed or calculated token expiry date.
319
320        Raises:
321            TypeError: If the result is not an instance of AirbyteDateTime.
322        """
323        expiry_date = self._get_config_value_by_path(self._token_expiry_date_config_path)
324        result = (
325            ab_datetime_now() - timedelta(days=1)
326            if expiry_date == ""
327            else ab_datetime_parse(str(expiry_date))
328        )
329        if isinstance(result, AirbyteDateTime):
330            return result
331        raise TypeError("Invalid datetime conversion")

Retrieves the token expiry date from the configuration.

This method fetches the token expiry date from the configuration using the specified path. If the expiry date is an empty string, it returns the current date and time minus one day. Otherwise, it parses the expiry date string into an AirbyteDateTime object.

Returns:

AirbyteDateTime: The parsed or calculated token expiry date.

Raises:
  • TypeError: If the result is not an instance of AirbyteDateTime.
def set_token_expiry_date( self, new_token_expiry_date: airbyte_cdk.utils.datetime_helpers.AirbyteDateTime) -> None: 
333    def set_token_expiry_date(self, new_token_expiry_date: AirbyteDateTime) -> None:  # type: ignore[override]
334        """
335        Sets the token expiry date in the configuration.
336
337        Args:
338            new_token_expiry_date (AirbyteDateTime): The new expiry date for the token.
339        """
340        if not (
341            isinstance(new_token_expiry_date, datetime) or self.token_expiry_is_time_of_expiration
342        ):
343            # Overrides of `refresh_access_token` written before 6.45.5 return the raw `expires_in`
344            # in seconds. Other values are stored as-is and parsed on read.
345            with suppress(ValueError, OverflowError):
346                new_token_expiry_date = self._parse_token_expiration_date(new_token_expiry_date)
347        self._set_config_value_by_path(
348            self._token_expiry_date_config_path, str(new_token_expiry_date)
349        )

Sets the token expiry date in the configuration.

Arguments:
  • new_token_expiry_date (AirbyteDateTime): The new expiry date for the token.
def token_has_expired(self) -> bool: 
351    def token_has_expired(self) -> bool:
352        """Returns True if the token is expired"""
353        return ab_datetime_now() > self.get_token_expiry_date()

Returns True if the token is expired

def get_access_token(self) -> str: 
355    def get_access_token(self) -> str:
356        """Retrieve new access and refresh token if the access token has expired.
357
358        This method uses double-checked locking to ensure thread-safe token refresh.
359        This is especially critical for single-use refresh tokens where concurrent
360        refresh attempts would cause failures as the refresh token is invalidated
361        after first use.
362
363        The new refresh token is persisted with the set_refresh_token function.
364
365        Returns:
366            str: The current access_token, updated if it was previously expired.
367        """
368        if self.token_has_expired():
369            with self._token_refresh_lock:
370                # Double-check after acquiring lock - another thread may have already refreshed
371                if self.token_has_expired():
372                    self.refresh_and_set_access_token()
373        return self.access_token

Retrieve new access and refresh token if the access token has expired.

This method uses double-checked locking to ensure thread-safe token refresh. This is especially critical for single-use refresh tokens where concurrent refresh attempts would cause failures as the refresh token is invalidated after first use.

The new refresh token is persisted with the set_refresh_token function.

Returns:

str: The current access_token, updated if it was previously expired.

def refresh_and_set_access_token(self) -> None: 
375    def refresh_and_set_access_token(self) -> None:
376        """Force refresh the access token and update internal state.
377
378        For single-use refresh tokens, this also persists the new refresh token
379        and emits a control message to update the connector config. If the
380        response omits a refresh token, the existing one is preserved.
381        """
382        token_before_waiting = self._current_access_token_or_none()
383        with self._token_refresh_lock:
384            if self._current_access_token_or_none() != token_before_waiting:
385                return
386            new_access_token, access_token_expires_in, new_refresh_token = (
387                self.refresh_access_token()
388            )
389            self.access_token = new_access_token
390            if new_refresh_token is not None:
391                self.set_refresh_token(new_refresh_token)
392            self.set_token_expiry_date(access_token_expires_in)
393            self._emit_control_message()

Force refresh the access token and update internal state.

For single-use refresh tokens, this also persists the new refresh token and emits a control message to update the connector config. If the response omits a refresh token, the existing one is preserved.

def refresh_access_token( self) -> Tuple[str, airbyte_cdk.utils.datetime_helpers.AirbyteDateTime, Optional[str]]: 
395    def refresh_access_token(self) -> Tuple[str, AirbyteDateTime, Optional[str]]:  # type: ignore[override]
396        """
397        Refreshes the access token by making a handled request and extracting the necessary token information.
398
399        Returns:
400            A tuple of (access_token, token_expiry_date, refresh_token). The refresh token
401            is `None` when the OAuth provider omits it from the response.
402        """
403        response_json = self._make_handled_request()
404        return (
405            self._extract_access_token(response_json),
406            self._extract_token_expiry_date(response_json),
407            self._extract_refresh_token(response_json),
408        )

Refreshes the access token by making a handled request and extracting the necessary token information.

Returns:

A tuple of (access_token, token_expiry_date, refresh_token). The refresh token is None when the OAuth provider omits it from the response.

Inherited Members
Oauth2Authenticator
get_token_refresh_endpoint
get_client_id_name
get_client_id
get_client_secret_name
get_client_secret
get_refresh_token_name
get_access_token_name
get_scopes
get_expires_in_name
get_refresh_request_body
get_refresh_request_headers
should_send_refresh_request_as_query_params
get_grant_type_name
get_grant_type
token_expiry_is_time_of_expiration
token_expiry_date_format
airbyte_cdk.sources.streams.http.requests_native_auth.abstract_oauth.AbstractOauth2Authenticator
get_auth_header
build_refresh_request_body
build_refresh_request_headers
build_refresh_request_query_params
class TokenAuthenticator(airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token.AbstractHeaderAuthenticator): 
39class TokenAuthenticator(AbstractHeaderAuthenticator):
40    """
41    Builds auth header, based on the token provided.
42    The token is attached to each request via the `auth_header` header.
43    """
44
45    @property
46    def auth_header(self) -> str:
47        return self._auth_header
48
49    @property
50    def token(self) -> str:
51        return f"{self._auth_method} {self._token}"
52
53    def __init__(self, token: str, auth_method: str = "Bearer", auth_header: str = "Authorization"):
54        self._auth_header = auth_header
55        self._auth_method = auth_method
56        self._token = token

Builds auth header, based on the token provided. The token is attached to each request via the auth_header header.

TokenAuthenticator( token: str, auth_method: str = 'Bearer', auth_header: str = 'Authorization') 
53    def __init__(self, token: str, auth_method: str = "Bearer", auth_header: str = "Authorization"):
54        self._auth_header = auth_header
55        self._auth_method = auth_method
56        self._token = token
auth_header: str 
45    @property
46    def auth_header(self) -> str:
47        return self._auth_header

HTTP header to set on the requests

token: str 
49    @property
50    def token(self) -> str:
51        return f"{self._auth_method} {self._token}"

The header value to set on outgoing HTTP requests

Inherited Members
airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token.AbstractHeaderAuthenticator
get_auth_header
class MultipleTokenAuthenticator(airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token.AbstractHeaderAuthenticator): 
15class MultipleTokenAuthenticator(AbstractHeaderAuthenticator):
16    """
17    Builds auth header, based on the list of tokens provided.
18    Auth header is changed per each `get_auth_header` call, using each token in cycle.
19    The token is attached to each request via the `auth_header` header.
20    """
21
22    @property
23    def auth_header(self) -> str:
24        return self._auth_header
25
26    @property
27    def token(self) -> str:
28        return f"{self._auth_method} {next(self._tokens_iter)}"
29
30    def __init__(
31        self, tokens: List[str], auth_method: str = "Bearer", auth_header: str = "Authorization"
32    ):
33        self._auth_method = auth_method
34        self._auth_header = auth_header
35        self._tokens = tokens
36        self._tokens_iter = cycle(self._tokens)

Builds auth header, based on the list of tokens provided. Auth header is changed per each get_auth_header call, using each token in cycle. The token is attached to each request via the auth_header header.

MultipleTokenAuthenticator( tokens: List[str], auth_method: str = 'Bearer', auth_header: str = 'Authorization') 
30    def __init__(
31        self, tokens: List[str], auth_method: str = "Bearer", auth_header: str = "Authorization"
32    ):
33        self._auth_method = auth_method
34        self._auth_header = auth_header
35        self._tokens = tokens
36        self._tokens_iter = cycle(self._tokens)
auth_header: str 
22    @property
23    def auth_header(self) -> str:
24        return self._auth_header

HTTP header to set on the requests

token: str 
26    @property
27    def token(self) -> str:
28        return f"{self._auth_method} {next(self._tokens_iter)}"

The header value to set on outgoing HTTP requests

Inherited Members
airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token.AbstractHeaderAuthenticator
get_auth_header
class BasicHttpAuthenticator(airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token.AbstractHeaderAuthenticator): 
59class BasicHttpAuthenticator(AbstractHeaderAuthenticator):
60    """
61    Builds auth based off the basic authentication scheme as defined by RFC 7617, which transmits credentials as USER ID/password pairs, encoded using bas64
62    https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication#basic_authentication_scheme
63    """
64
65    @property
66    def auth_header(self) -> str:
67        return self._auth_header
68
69    @property
70    def token(self) -> str:
71        return f"{self._auth_method} {self._token}"
72
73    def __init__(
74        self,
75        username: str,
76        password: str = "",
77        auth_method: str = "Basic",
78        auth_header: str = "Authorization",
79    ):
80        auth_string = f"{username}:{password}".encode("utf8")
81        b64_encoded = base64.b64encode(auth_string).decode("utf8")
82        self._auth_header = auth_header
83        self._auth_method = auth_method
84        self._token = b64_encoded

Builds auth based off the basic authentication scheme as defined by RFC 7617, which transmits credentials as USER ID/password pairs, encoded using bas64 https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication#basic_authentication_scheme

BasicHttpAuthenticator( username: str, password: str = '', auth_method: str = 'Basic', auth_header: str = 'Authorization') 
73    def __init__(
74        self,
75        username: str,
76        password: str = "",
77        auth_method: str = "Basic",
78        auth_header: str = "Authorization",
79    ):
80        auth_string = f"{username}:{password}".encode("utf8")
81        b64_encoded = base64.b64encode(auth_string).decode("utf8")
82        self._auth_header = auth_header
83        self._auth_method = auth_method
84        self._token = b64_encoded
auth_header: str 
65    @property
66    def auth_header(self) -> str:
67        return self._auth_header

HTTP header to set on the requests

token: str 
69    @property
70    def token(self) -> str:
71        return f"{self._auth_method} {self._token}"

The header value to set on outgoing HTTP requests

Inherited Members
airbyte_cdk.sources.streams.http.requests_native_auth.abstract_token.AbstractHeaderAuthenticator
get_auth_header
@runtime_checkable
class ResponseAwareAuthenticator(typing.Protocol): 
30@runtime_checkable
31class ResponseAwareAuthenticator(Protocol):
32    """An authenticator that wants to see responses, not just sign requests.
33
34    Authenticators tracking per-token quota need a feedback channel: without one they can only
35    guess at the server's view of the quota and cannot tell that a token has been rejected.
36    `HttpClient` calls `update_from_response` once per attempt on any authenticator that
37    implements this, skipping replayed cache hits -- a cached response carries stale rate-limit
38    headers and consumed no quota.
39
40    Implementations must not raise for a response they cannot interpret, and must be safe to
41    call from multiple threads.
42    """
43
44    def update_from_response(
45        self, request: requests.PreparedRequest, response: requests.Response
46    ) -> None:
47        pass

An authenticator that wants to see responses, not just sign requests.

Authenticators tracking per-token quota need a feedback channel: without one they can only guess at the server's view of the quota and cannot tell that a token has been rejected. HttpClient calls update_from_response once per attempt on any authenticator that implements this, skipping replayed cache hits -- a cached response carries stale rate-limit headers and consumed no quota.

Implementations must not raise for a response they cannot interpret, and must be safe to call from multiple threads.

ResponseAwareAuthenticator(*args, **kwargs) 
1431def _no_init_or_replace_init(self, *args, **kwargs):
1432    cls = type(self)
1433
1434    if cls._is_protocol:
1435        raise TypeError('Protocols cannot be instantiated')
1436
1437    # Already using a custom `__init__`. No need to calculate correct
1438    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1439    if cls.__init__ is not _no_init_or_replace_init:
1440        return
1441
1442    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1443    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1444    # searches for a proper new `__init__` in the MRO. The new `__init__`
1445    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1446    # instantiation of the protocol subclass will thus use the new
1447    # `__init__` and no longer call `_no_init_or_replace_init`.
1448    for base in cls.__mro__:
1449        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1450        if init is not _no_init_or_replace_init:
1451            cls.__init__ = init
1452            break
1453    else:
1454        # should not happen
1455        cls.__init__ = object.__init__
1456
1457    cls.__init__(self, *args, **kwargs)
def update_from_response( self, request: requests.models.PreparedRequest, response: requests.models.Response) -> None: 
44    def update_from_response(
45        self, request: requests.PreparedRequest, response: requests.Response
46    ) -> None:
47        pass
@runtime_checkable
class TokenRotatingAuthenticator(typing.Protocol): 
50@runtime_checkable
51class TokenRotatingAuthenticator(Protocol):
52    """An authenticator holding several interchangeable credentials.
53
54    A rate-limit backoff is computed from the response alone, so it assumes the only way
55    forward is for that quota to come back. An authenticator with a spare credential knows
56    better. `HttpClient` asks before sleeping out a rate-limit window.
57
58    Implementations must only answer True when retrying immediately would actually use a
59    different credential -- otherwise the retry hammers the same rejected one.
60    """
61
62    def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
63        pass

An authenticator holding several interchangeable credentials.

A rate-limit backoff is computed from the response alone, so it assumes the only way forward is for that quota to come back. An authenticator with a spare credential knows better. HttpClient asks before sleeping out a rate-limit window.

Implementations must only answer True when retrying immediately would actually use a different credential -- otherwise the retry hammers the same rejected one.

TokenRotatingAuthenticator(*args, **kwargs) 
1431def _no_init_or_replace_init(self, *args, **kwargs):
1432    cls = type(self)
1433
1434    if cls._is_protocol:
1435        raise TypeError('Protocols cannot be instantiated')
1436
1437    # Already using a custom `__init__`. No need to calculate correct
1438    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1439    if cls.__init__ is not _no_init_or_replace_init:
1440        return
1441
1442    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1443    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1444    # searches for a proper new `__init__` in the MRO. The new `__init__`
1445    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1446    # instantiation of the protocol subclass will thus use the new
1447    # `__init__` and no longer call `_no_init_or_replace_init`.
1448    for base in cls.__mro__:
1449        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1450        if init is not _no_init_or_replace_init:
1451            cls.__init__ = init
1452            break
1453    else:
1454        # should not happen
1455        cls.__init__ = object.__init__
1456
1457    cls.__init__(self, *args, **kwargs)
def has_alternative_token(self, request: requests.models.PreparedRequest) -> bool: 
62    def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
63        pass