Merge branch 'master' into ddavydov/#1313-source-google-ads-write-few…

…er-logs
airbytehq · Jan 20, 2023 · 0b61eb3 · 0b61eb3
2 parents 1d995cf + de91191
commit 0b61eb3
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 2 deletions.
diff --git a/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -359,7 +359,7 @@ def create_declarative_stream(model: DeclarativeStreamModel, config: Config, **k
     if model.schema_loader:
         schema_loader = _create_component_from_model(model=model.schema_loader, config=config)
     else:
-        schema_loader = DefaultSchemaLoader(config=config, options=model.options)
+        schema_loader = DefaultSchemaLoader(config=config, name=model.name, options=model.options)
 
     transformations = []
     if model.transformations:

diff --git a/airbyte-cdk/python/airbyte_cdk/sources/declarative/schema/default_schema_loader.py b/airbyte-cdk/python/airbyte_cdk/sources/declarative/schema/default_schema_loader.py
@@ -23,9 +23,13 @@ class DefaultSchemaLoader(SchemaLoader, JsonSchemaMixin):
     """
 
     config: Config
+    name: InitVar[str]
     options: InitVar[Mapping[str, Any]]
 
-    def __post_init__(self, options: Mapping[str, Any]):
+    def __post_init__(self, name: str, options: Mapping[str, Any]):
+        options = options or {}
+        if "name" not in options:
+            options["name"] = name
         self._options = options
         self.default_loader = JsonFileSchemaLoader(options=options, config=self.config)
 

diff --git a/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
@@ -1011,3 +1011,37 @@ def test_add_fields(self):
             )
         ]
         assert stream.transformations == expected
+
+    def test_default_schema_loader(self):
+        component_definition = {
+            "type": "DeclarativeStream",
+            "name": "test",
+            "primary_key": [],
+            "retriever": {
+                "type": "SimpleRetriever",
+                "name": "test",
+                "primary_key": [],
+                "requester": {
+                    "type": "HttpRequester",
+                    "name": "test",
+                    "url_base": "http://localhost:6767/",
+                    "path": "items/",
+                    "request_options_provider": {
+                        "request_parameters": {},
+                        "request_headers": {},
+                        "request_body_json": {},
+                        "type": "InterpolatedRequestOptionsProvider",
+                    },
+                    "authenticator": {"type": "BearerAuthenticator", "api_token": "{{ config['api_key'] }}"},
+                },
+                "record_selector": {"type": "RecordSelector", "extractor": {"type": "DpathExtractor", "field_pointer": ["items"]}},
+                "paginator": {"type": "NoPagination"},
+            },
+        }
+        resolved_manifest = resolver.preprocess_manifest(component_definition)
+        propagated_source_config = ManifestComponentTransformer().propagate_types_and_options("", resolved_manifest, {})
+        stream = factory.create_component(
+            model_type=DeclarativeStreamModel, component_definition=propagated_source_config, config=input_config
+        )
+        schema_loader = stream.schema_loader
+        assert schema_loader.default_loader._get_json_filepath() == f"./{stream.name}.json"