Airbyte CDK (low code): add refresh_token_error handler to `Declarati…

…veOauth2Authenticator` (#36058)

Signed-off-by: Artem Inzhyyants <artem.inzhyyants@gmail.com>
Co-authored-by: Alexandre Girard <alexandre@airbyte.io>

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -924,6 +924,31 @@ definitions:
             default: ["credentials", "token_expiry_date"]
             examples:
               - ["credentials", "token_expiry_date"]
+          refresh_token_error_status_codes:
+            title: Refresh Token Error Status Codes
+            description: Status Codes to Identify refresh token error in response (Refresh Token Error Key and Refresh Token Error Values should be also specified). Responses with one of the error status code and containing an error value will be flagged as a config error
+            type: array
+            items:
+              type: integer
+            default: []
+            examples:
+              - [400, 500]
+          refresh_token_error_key:
+            title: Refresh Token Error Key
+            description: Key to Identify refresh token error in response (Refresh Token Error Status Codes and Refresh Token Error Values should be also specified).
+            type: string
+            default: ""
+            examples:
+              - "error"
+          refresh_token_error_values:
+            title: Refresh Token Error Values
+            description: 'List of values to check for exception during token refresh process. Used to check if the error found in the response matches the key from the Refresh Token Error Key field (e.g. response={"error": "invalid_grant"}). Only responses with one of the error status code and containing an error value will be flagged as a config error'
+            type: array
+            items:
+              type: string
+            default: []
+            examples:
+              - ["invalid_grant", "invalid_permissions"]
       $parameters:
         type: object
         additionalProperties: true

airbyte-cdk/python/airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -261,6 +261,24 @@ class RefreshTokenUpdater(BaseModel):
         examples=[['credentials', 'token_expiry_date']],
         title='Config Path To Expiry Date',
     )
+    refresh_token_error_status_codes: Optional[List[int]] = Field(
+        [],
+        description='Status Codes to Identify refresh token error in response (Refresh Token Error Key and Refresh Token Error Values should be also specified). Responses with one of the error status code and containing an error value will be flagged as a config error',
+        examples=[[400, 500]],
+        title='Refresh Token Error Status Codes',
+    )
+    refresh_token_error_key: Optional[str] = Field(
+        '',
+        description='Key to Identify refresh token error in response (Refresh Token Error Status Codes and Refresh Token Error Values should be also specified).',
+        examples=['error'],
+        title='Refresh Token Error Key',
+    )
+    refresh_token_error_values: Optional[List[str]] = Field(
+        [],
+        description='List of values to check for exception during token refresh process. Used to check if the error found in the response matches the key from the Refresh Token Error Key field (e.g. response={"error": "invalid_grant"}). Only responses with one of the error status code and containing an error value will be flagged as a config error',
+        examples=[['invalid_grant', 'invalid_permissions']],
+        title='Refresh Token Error Values',
+    )
 
 
 class OAuthAuthenticator(BaseModel):

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -808,7 +808,7 @@ def create_no_pagination(model: NoPaginationModel, config: Config, **kwargs: Any
 
     def create_oauth_authenticator(self, model: OAuthAuthenticatorModel, config: Config, **kwargs: Any) -> DeclarativeOauth2Authenticator:
         if model.refresh_token_updater:
-            # ignore type error beause fixing it would have a lot of dependencies, revisit later
+            # ignore type error because fixing it would have a lot of dependencies, revisit later
             return DeclarativeSingleUseRefreshTokenOauth2Authenticator(  # type: ignore
                 config,
                 InterpolatedString.create(model.token_refresh_endpoint, parameters=model.parameters or {}).eval(config),
@@ -829,6 +829,9 @@ def create_oauth_authenticator(self, model: OAuthAuthenticatorModel, config: Con
                 scopes=model.scopes,
                 token_expiry_date_format=model.token_expiry_date_format,
                 message_repository=self._message_repository,
+                refresh_token_error_status_codes=model.refresh_token_updater.refresh_token_error_status_codes,
+                refresh_token_error_key=model.refresh_token_updater.refresh_token_error_key,
+                refresh_token_error_values=model.refresh_token_updater.refresh_token_error_values,
             )
         # ignore type error because fixing it would have a lot of dependencies, revisit later
         return DeclarativeOauth2Authenticator(  # type: ignore

airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -359,6 +359,9 @@ def test_single_use_oauth_branch():
         interpolated_body_field: "{{ config['apikey'] }}"
       refresh_token_updater:
         refresh_token_name: "the_refresh_token"
+        refresh_token_error_status_codes: [400]
+        refresh_token_error_key: "error"
+        refresh_token_error_values: ["invalid_grant"]
         refresh_token_config_path:
           - apikey
     """
@@ -381,6 +384,9 @@ def test_single_use_oauth_branch():
     # default values
     assert authenticator._access_token_config_path == ["credentials", "access_token"]
     assert authenticator._token_expiry_date_config_path == ["credentials", "token_expiry_date"]
+    assert authenticator._refresh_token_error_status_codes == [400]
+    assert authenticator._refresh_token_error_key == "error"
+    assert authenticator._refresh_token_error_values == ["invalid_grant"]
 
 
 def test_list_based_stream_slicer_with_values_refd():