Merge main to branch

Author: lazebnyi

airbyte_cdk/sources/declarative/checks/__init__.py

@@ -7,7 +7,10 @@
 from pydantic.v1 import BaseModel
 
 from airbyte_cdk.sources.declarative.checks.check_dynamic_stream import CheckDynamicStream
-from airbyte_cdk.sources.declarative.checks.check_stream import CheckStream, DynamicStreamCheckConfig
+from airbyte_cdk.sources.declarative.checks.check_stream import (
+    CheckStream,
+    DynamicStreamCheckConfig,
+)
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 from airbyte_cdk.sources.declarative.models import (
     CheckDynamicStream as CheckDynamicStreamModel,

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -611,9 +611,7 @@ class OAuthAuthenticator(BaseModel):
     scopes: Optional[List[str]] = Field(
         None,
         description="List of scopes that should be granted to the access token.",
-        examples=[
-            ["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"]
-        ],
+        examples=[["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"]],
         title="Scopes",
     )
     token_expiry_date: Optional[str] = Field(
@@ -1087,28 +1085,24 @@ class OAuthConfigSpecification(BaseModel):
     class Config:
         extra = Extra.allow
 
-    oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = (
-        Field(
-            None,
-            description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['app_id']\n    }\n  }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['info', 'app_id']\n    }\n  }",
-            examples=[
-                {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}},
-                {
-                    "app_id": {
-                        "type": "string",
-                        "path_in_connector_config": ["info", "app_id"],
-                    }
-                },
-            ],
-            title="OAuth user input",
-        )
+    oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = Field(
+        None,
+        description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['app_id']\n    }\n  }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['info', 'app_id']\n    }\n  }",
+        examples=[
+            {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}},
+            {
+                "app_id": {
+                    "type": "string",
+                    "path_in_connector_config": ["info", "app_id"],
+                }
+            },
+        ],
+        title="OAuth user input",
     )
-    oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = (
-        Field(
-            None,
-            description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n  + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n  + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n  + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n  + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n  + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n  - The TikTok Marketing DeclarativeOAuth spec:\n  {\n    "oauth_connector_input_specification": {\n      "type": "object",\n      "additionalProperties": false,\n      "properties": {\n          "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n          "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n          "access_token_params": {\n              "{{ auth_code_key }}": "{{ auth_code_value }}",\n              "{{ client_id_key }}": "{{ client_id_value }}",\n              "{{ client_secret_key }}": "{{ client_secret_value }}"\n          },\n          "access_token_headers": {\n              "Content-Type": "application/json",\n              "Accept": "application/json"\n          },\n          "extract_output": ["data.access_token"],\n          "client_id_key": "app_id",\n          "client_secret_key": "secret",\n          "auth_code_key": "auth_code"\n      }\n    }\n  }',
-            title="DeclarativeOAuth Connector Specification",
-        )
+    oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = Field(
+        None,
+        description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n  + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n  + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n  + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n  + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n  + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n  - The TikTok Marketing DeclarativeOAuth spec:\n  {\n    "oauth_connector_input_specification": {\n      "type": "object",\n      "additionalProperties": false,\n      "properties": {\n          "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n          "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n          "access_token_params": {\n              "{{ auth_code_key }}": "{{ auth_code_value }}",\n              "{{ client_id_key }}": "{{ client_id_value }}",\n              "{{ client_secret_key }}": "{{ client_secret_value }}"\n          },\n          "access_token_headers": {\n              "Content-Type": "application/json",\n              "Accept": "application/json"\n          },\n          "extract_output": ["data.access_token"],\n          "client_id_key": "app_id",\n          "client_secret_key": "secret",\n          "auth_code_key": "auth_code"\n      }\n    }\n  }',
+        title="DeclarativeOAuth Connector Specification",
     )
     complete_oauth_output_specification: Optional[Dict[str, Any]] = Field(
         None,
@@ -1126,9 +1120,7 @@ class Config:
     complete_oauth_server_input_specification: Optional[Dict[str, Any]] = Field(
         None,
         description="OAuth specific blob. This is a Json Schema used to validate Json configurations persisted as Airbyte Server configurations.\nMust be a valid non-nested JSON describing additional fields configured by the Airbyte Instance or Workspace Admins to be used by the\nserver when completing an OAuth flow (typically exchanging an auth code for refresh token).\nExamples:\n    complete_oauth_server_input_specification={\n      client_id: {\n        type: string\n      },\n      client_secret: {\n        type: string\n      }\n    }",
-        examples=[
-            {"client_id": {"type": "string"}, "client_secret": {"type": "string"}}
-        ],
+        examples=[{"client_id": {"type": "string"}, "client_secret": {"type": "string"}}],
         title="OAuth input specification",
     )
     complete_oauth_server_output_specification: Optional[Dict[str, Any]] = Field(
@@ -1802,9 +1794,7 @@ class RecordSelector(BaseModel):
         description="Responsible for filtering records to be emitted by the Source.",
         title="Record Filter",
     )
-    schema_normalization: Optional[
-        Union[SchemaNormalization, CustomSchemaNormalization]
-    ] = Field(
+    schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field(
         SchemaNormalization.None_,
         description="Responsible for normalization according to the schema.",
         title="Schema Normalization",
@@ -2031,9 +2021,7 @@ class Config:
         description="Component used to fetch data incrementally based on a time field in the data.",
         title="Incremental Sync",
     )
-    name: Optional[str] = Field(
-        "", description="The stream name.", example=["Users"], title="Name"
-    )
+    name: Optional[str] = Field("", description="The stream name.", example=["Users"], title="Name")
     primary_key: Optional[PrimaryKey] = Field(
         "", description="The primary key of the stream.", title="Primary Key"
     )
@@ -2291,9 +2279,7 @@ class ParentStreamConfig(BaseModel):
 
 class StateDelegatingStream(BaseModel):
     type: Literal["StateDelegatingStream"]
-    name: str = Field(
-        ..., description="The stream name.", example=["Users"], title="Name"
-    )
+    name: str = Field(..., description="The stream name.", example=["Users"], title="Name")
     full_refresh_stream: DeclarativeStream = Field(
         ...,
         description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.",
@@ -2382,9 +2368,7 @@ class AsyncRetriever(BaseModel):
     )
     download_extractor: Optional[
         Union[CustomRecordExtractor, DpathExtractor, ResponseToFileExtractor]
-    ] = Field(
-        None, description="Responsible for fetching the records from provided urls."
-    )
+    ] = Field(None, description="Responsible for fetching the records from provided urls.")
     creation_requester: Union[CustomRequester, HttpRequester] = Field(
         ...,
         description="Requester component that describes how to prepare HTTP requests to send to the source API to create the async server-side job.",
@@ -2524,12 +2508,10 @@ class DynamicDeclarativeStream(BaseModel):
     stream_template: DeclarativeStream = Field(
         ..., description="Reference to the stream template.", title="Stream Template"
     )
-    components_resolver: Union[HttpComponentsResolver, ConfigComponentsResolver] = (
-        Field(
-            ...,
-            description="Component resolve and populates stream templates with components values.",
-            title="Components Resolver",
-        )
+    components_resolver: Union[HttpComponentsResolver, ConfigComponentsResolver] = Field(
+        ...,
+        description="Component resolve and populates stream templates with components values.",
+        title="Components Resolver",
     )
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -54,7 +54,11 @@
     SessionTokenProvider,
     TokenProvider,
 )
-from airbyte_cdk.sources.declarative.checks import CheckDynamicStream, CheckStream, DynamicStreamCheckConfig
+from airbyte_cdk.sources.declarative.checks import (
+    CheckDynamicStream,
+    CheckStream,
+    DynamicStreamCheckConfig,
+)
 from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
 from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.declarative_stream import DeclarativeStream
@@ -131,9 +135,6 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     CheckStream as CheckStreamModel,
 )
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    DynamicStreamCheckConfig as DynamicStreamCheckConfigModel,
-)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ComplexFieldType as ComplexFieldTypeModel,
 )
@@ -221,6 +222,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DynamicSchemaLoader as DynamicSchemaLoaderModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    DynamicStreamCheckConfig as DynamicStreamCheckConfigModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ExponentialBackoffStrategy as ExponentialBackoffStrategyModel,
 )
@@ -941,25 +945,35 @@ def create_bearer_authenticator(
 
     @staticmethod
     def create_dynamic_stream_check_config(
-            model: DynamicStreamCheckConfigModel, config: Config, **kwargs: Any
+        model: DynamicStreamCheckConfigModel, config: Config, **kwargs: Any
     ) -> DynamicStreamCheckConfig:
         return DynamicStreamCheckConfig(
             dynamic_stream_name=model.dynamic_stream_name,
             stream_count=model.stream_count or 0,
         )
 
-    def create_check_stream(self, model: CheckStreamModel, config: Config, **kwargs: Any) -> CheckStream:
+    def create_check_stream(
+        self, model: CheckStreamModel, config: Config, **kwargs: Any
+    ) -> CheckStream:
         if model.dynamic_streams_check_configs is None and model.stream_names is None:
             raise ValueError(
                 "Expected either stream_names or dynamic_streams_check_configs to be set for CheckStream"
             )
 
-        dynamic_streams_check_configs = [
-            self._create_component_from_model(model=dynamic_stream_check_config, config=config)
-            for dynamic_stream_check_config in model.dynamic_streams_check_configs
-        ] if model.dynamic_streams_check_configs else []
+        dynamic_streams_check_configs = (
+            [
+                self._create_component_from_model(model=dynamic_stream_check_config, config=config)
+                for dynamic_stream_check_config in model.dynamic_streams_check_configs
+            ]
+            if model.dynamic_streams_check_configs
+            else []
+        )
 
-        return CheckStream(stream_names=model.stream_names or [], dynamic_streams_check_configs=dynamic_streams_check_configs, parameters={})
+        return CheckStream(
+            stream_names=model.stream_names or [],
+            dynamic_streams_check_configs=dynamic_streams_check_configs,
+            parameters={},
+        )
 
     @staticmethod
     def create_check_dynamic_stream(

unit_tests/sources/declarative/checks/test_check_stream.py

@@ -4,19 +4,19 @@
 
 import json
 import logging
+from copy import deepcopy
 from typing import Any, Iterable, Mapping, Optional
 from unittest.mock import MagicMock
-from copy import deepcopy
 
 import pytest
 import requests
 
 from jsonschema.exceptions import ValidationError
 from airbyte_cdk.sources.declarative.checks.check_stream import CheckStream
-from airbyte_cdk.sources.streams.http import HttpStream
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
 )
+from airbyte_cdk.sources.streams.http import HttpStream
 from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
 
 logger = logging.getLogger("test")
@@ -31,19 +31,19 @@
     [
         ("test_success_check", record, stream_names, {}, (True, None)),
         (
-                "test_success_check_stream_slice",
-                record,
-                stream_names,
-                {"slice": "slice_value"},
-                (True, None),
+            "test_success_check_stream_slice",
+            record,
+            stream_names,
+            {"slice": "slice_value"},
+            (True, None),
         ),
         ("test_fail_check", None, stream_names, {}, (True, None)),
         ("test_try_to_check_invalid stream", record, ["invalid_stream_name"], {}, None),
     ],
 )
 @pytest.mark.parametrize("slices_as_list", [True, False])
 def test_check_stream_with_slices_as_list(
-        test_name, record, streams_to_check, stream_slice, expectation, slices_as_list
+    test_name, record, streams_to_check, stream_slice, expectation, slices_as_list
 ):
     stream = MagicMock()
     stream.name = "s1"
@@ -108,22 +108,22 @@ def test_check_stream_with_no_stream_slices_aborts():
     "test_name, response_code, available_expectation, expected_messages",
     [
         (
-                "test_stream_unavailable_unhandled_error",
-                404,
-                False,
-                ["Not found. The requested resource was not found on the server."],
+            "test_stream_unavailable_unhandled_error",
+            404,
+            False,
+            ["Not found. The requested resource was not found on the server."],
         ),
         (
-                "test_stream_unavailable_handled_error",
-                403,
-                False,
-                ["Forbidden. You don't have permission to access this resource."],
+            "test_stream_unavailable_handled_error",
+            403,
+            False,
+            ["Forbidden. You don't have permission to access this resource."],
         ),
         ("test_stream_available", 200, True, []),
     ],
 )
 def test_check_http_stream_via_availability_strategy(
-        mocker, test_name, response_code, available_expectation, expected_messages
+    mocker, test_name, response_code, available_expectation, expected_messages
 ):
     class MockHttpStream(HttpStream):
         url_base = "https://test_base_url.com"
@@ -166,10 +166,13 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
         assert message in reason
 
 
-_CONFIG = {"start_date": "2024-07-01T00:00:00.000Z", "custom_streams": [
-    {"id": 3, "name": "item_3"},
-    {"id": 4, "name": "item_4"},
-]}
+_CONFIG = {
+    "start_date": "2024-07-01T00:00:00.000Z",
+    "custom_streams": [
+        {"id": 3, "name": "item_3"},
+        {"id": 4, "name": "item_4"},
+    ],
+}
 
 _MANIFEST_WITHOUT_CHECK_COMPONENT = {
     "version": "6.7.0",
@@ -316,7 +319,7 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
                     },
                 ],
             },
-        }
+        },
     ],
     "streams": [
         {
@@ -355,7 +358,7 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
                 },
             }
         }
-    ]
+    ],
 }