TMP cdk features for the release of bing-ads to low-code

airbyte_cdk/sources/declarative/auth/oauth.py

@@ -1,10 +1,10 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-
+import logging
 from dataclasses import InitVar, dataclass, field
 from datetime import datetime, timedelta
-from typing import Any, List, Mapping, MutableMapping, Optional, Union
+from typing import Any, List, Mapping, Optional, Union
 
 from airbyte_cdk.sources.declarative.auth.declarative_authenticator import DeclarativeAuthenticator
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
@@ -19,6 +19,8 @@
 )
 from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
+logger = logging.getLogger("airbyte")
+
 
 @dataclass
 class DeclarativeOauth2Authenticator(AbstractOauth2Authenticator, DeclarativeAuthenticator):
@@ -201,7 +203,8 @@ def get_client_secret(self) -> str:
             self._client_secret.eval(self.config) if self._client_secret else self._client_secret
         )
         if not client_secret:
-            raise ValueError("OAuthAuthenticator was unable to evaluate client_secret parameter")
+            # We've seen some APIs allowing empty client_secret so we will only log here
+            logger.warning("OAuthAuthenticator was unable to evaluate client_secret parameter hence it'll be empty")
         return client_secret  # type: ignore # value will be returned as a string, or an error will be raised
 
     def get_refresh_token_name(self) -> str:

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -88,6 +88,7 @@ def __init__(
             emit_connector_builder_messages=emit_connector_builder_messages,
             disable_resumable_full_refresh=True,
             connector_state_manager=self._connector_state_manager,
+            max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
         )
 
         super().__init__(

airbyte_cdk/sources/declarative/requesters/http_job_repository.py

@@ -320,14 +320,13 @@ def _get_polling_response_interpolation_context(self, job: AsyncJob) -> Dict[str
         return polling_response_context
 
     def _get_create_job_stream_slice(self, job: AsyncJob) -> StreamSlice:
-        stream_slice = StreamSlice(
-            partition={},
-            cursor_slice={},
-            extra_fields={
+        return StreamSlice(
+            partition=job.job_parameters().partition,
+            cursor_slice=job.job_parameters().cursor_slice,
+            extra_fields=dict(job.job_parameters().extra_fields) | {
                 "creation_response": self._get_creation_response_interpolation_context(job),
             },
         )
-        return stream_slice
 
     def _get_download_targets(self, job: AsyncJob) -> Iterable[str]:
         if not self.download_target_requester:

airbyte_cdk/sources/declarative/requesters/request_options/interpolated_nested_request_input_provider.py

@@ -11,6 +11,7 @@
 )
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.types import Config, StreamSlice
+from airbyte_cdk.utils.mapping_helpers import get_interpolation_context
 
 
 @dataclass
@@ -52,8 +53,8 @@ def eval_request_inputs(
         :param next_page_token: The pagination token
         :return: The request inputs to set on an outgoing HTTP request
         """
-        kwargs = {
-            "stream_slice": stream_slice,
-            "next_page_token": next_page_token,
-        }
+        kwargs = get_interpolation_context(
+            stream_slice=stream_slice,
+            next_page_token=next_page_token,
+        )
         return self._interpolator.eval(self.config, **kwargs)  # type: ignore  # self._interpolator is always initialized with a value and will not be None

airbyte_cdk/sources/declarative/requesters/request_options/interpolated_request_input_provider.py

@@ -8,6 +8,7 @@
 from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
+from airbyte_cdk.utils.mapping_helpers import get_interpolation_context
 
 
 @dataclass
@@ -51,10 +52,10 @@ def eval_request_inputs(
         :param valid_value_types: A tuple of types that the interpolator should allow
         :return: The request inputs to set on an outgoing HTTP request
         """
-        kwargs = {
-            "stream_slice": stream_slice,
-            "next_page_token": next_page_token,
-        }
+        kwargs = get_interpolation_context(
+            stream_slice=stream_slice,
+            next_page_token=next_page_token,
+        )
         interpolated_value = self._interpolator.eval(  # type: ignore # self._interpolator is always initialized with a value and will not be None
             self.config,
             valid_key_types=valid_key_types,

unit_tests/sources/declarative/auth/test_oauth.py

@@ -5,6 +5,7 @@
 import base64
 import json
 import logging
+from copy import deepcopy
 from datetime import timedelta, timezone
 from unittest.mock import Mock
 
@@ -128,6 +129,20 @@ def test_refresh_with_encode_config_params(self):
         }
         assert body == expected
 
+    def test_client_secret_empty(self):
+        config_without_client_secret = deepcopy(config)
+        del config_without_client_secret["client_secret"]
+        oauth = DeclarativeOauth2Authenticator(
+            token_refresh_endpoint="{{ config['refresh_endpoint'] }}",
+            client_id="{{ config['client_id'] }}",
+            client_secret="{{ config['client_secret'] }}",
+            config=config_without_client_secret,
+            parameters={},
+            grant_type="client_credentials",
+        )
+        body = oauth.build_refresh_request_body()
+        assert body["client_secret"] == ""
+
     def test_refresh_with_decode_config_params(self):
         updated_config_fields = {
             "client_id": base64.b64encode(config["client_id"].encode("utf-8")).decode(),

unit_tests/sources/declarative/parsers/conftest.py

@@ -1,7 +1,7 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-
+from pathlib import Path
 from typing import Any, Dict
 
 import pytest
@@ -645,10 +645,7 @@ def expected_manifest_with_url_base_linked_definition_normalized() -> Dict[str,
 
 @pytest.fixture
 def manifest_with_linked_definitions_url_base_authenticator_abnormal_schemas() -> Dict[str, Any]:
-    with open(
-        "unit_tests/sources/declarative/parsers/resources/abnormal_schemas_manifest.yaml",
-        "r",
-    ) as file:
+    with open(str(Path(__file__).parent / "resources/abnormal_schemas_manifest.yaml"), "r") as file:
         return dict(yaml.safe_load(file))