feat(airbyte-cdk): add json_schema from ConfiguredCatalog to Stream

[Artem Inzhyyants (artem1205)]

What

Resolving https://github.com/airbytehq/airbyte-internal-issues/issues/2100

How

add json_schema from configured catalog to Stream

Review guide

1. airbyte-cdk/python/airbyte_cdk/sources/streams/core.py
2. airbyte-cdk/python/unit_tests/sources/streams/test_stream_read.py

User Impact

Can this PR be safely reverted and rolled back?

• YES 💚
• NO ❌

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Jun 18, 2024 9:06am

[Artem Inzhyyants (artem1205)]

Changes it codebase (source marketo)

Index: airbyte-integrations/connectors/source-marketo/integration_tests/configured_catalog.json
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/airbyte-integrations/connectors/source-marketo/integration_tests/configured_catalog.json b/airbyte-integrations/connectors/source-marketo/integration_tests/configured_catalog.json
--- a/airbyte-integrations/connectors/source-marketo/integration_tests/configured_catalog.json	(revision ddf755c0c86de3f6e3b3008d9cfb3eb87056864b)
+++ b/airbyte-integrations/connectors/source-marketo/integration_tests/configured_catalog.json	(date 1718628420139)
@@ -1,51 +1,32 @@
 {
   "streams": [
     {
-      "stream": {
-        "name": "programs",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh", "incremental"],
-        "source_defined_cursor": true,
-        "default_cursor_field": ["updatedAt"],
-        "source_defined_primary_key": [["id"]]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "append",
-      "cursor_field": ["updatedAt"],
-      "primary_key": [["id"]]
-    },
-    {
-      "stream": {
-        "name": "campaigns",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh", "incremental"],
-        "source_defined_cursor": true,
-        "default_cursor_field": ["createdAt"],
-        "source_defined_primary_key": [["id"]]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "append",
-      "cursor_field": ["createdAt"],
-      "primary_key": [["id"]]
-    },
-    {
-      "stream": {
-        "name": "lists",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh", "incremental"],
-        "source_defined_cursor": true,
-        "default_cursor_field": ["createdAt"],
-        "source_defined_primary_key": [["id"]]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "append",
-      "cursor_field": ["createdAt"],
-      "primary_key": [["id"]]
-    },
-    {
       "stream": {
         "name": "leads",
-        "json_schema": {},
+        "json_schema": {
+          "type": ["object", "null"],
+          "additionalProperties": true,
+          "properties": {
+            "company": {
+              "description": "The name of the company associated with the lead.",
+              "type": ["string", "null"]
+            },
+            "site": {
+              "description": "The website associated with the lead.",
+              "type": ["string", "null"]
+            },
+            "id": {
+              "description": "The unique identifier of the lead.",
+              "type": ["integer", "null"]
+            },
+            "updatedAt": {
+              "description": "The date and time when the lead was last updated.",
+              "type": ["string", "null"],
+              "format": "date-time"
+            }
+          }}
+,
+
         "supported_sync_modes": ["full_refresh", "incremental"],
         "default_cursor_field": ["updatedAt"],
         "source_defined_primary_key": [["id"]]
@@ -54,43 +35,6 @@
       "destination_sync_mode": "append",
       "cursor_field": ["updatedAt"],
       "primary_key": [["id"]]
-    },
-    {
-      "stream": {
-        "name": "activity_types",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"],
-        "source_defined_cursor": false,
-        "default_cursor_field": null,
-        "source_defined_primary_key": [["id"]]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "append",
-      "cursor_field": null,
-      "primary_key": [["id"]]
-    },
-    {
-      "stream": {
-        "name": "activities_visit_webpage",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh", "incremental"],
-        "source_defined_cursor": true,
-        "default_cursor_field": ["activityDate"],
-        "source_defined_primary_key": [["marketoGUID"]]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "append",
-      "cursor_field": ["activityDate"],
-      "primary_key": [["marketoGUID"]]
-    },
-    {
-      "stream": {
-        "name": "segmentations",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite"
     }
   ]
 }
Index: airbyte-integrations/connectors/source-marketo/source_marketo/source.py
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/airbyte-integrations/connectors/source-marketo/source_marketo/source.py b/airbyte-integrations/connectors/source-marketo/source_marketo/source.py
--- a/airbyte-integrations/connectors/source-marketo/source_marketo/source.py	(revision ddf755c0c86de3f6e3b3008d9cfb3eb87056864b)
+++ b/airbyte-integrations/connectors/source-marketo/source_marketo/source.py	(date 1718628542439)
@@ -364,16 +364,12 @@
 
     @property
     def stream_fields(self):
-        standard_properties = set(self.get_json_schema()["properties"])
-        resp = self._session.get(f"{self._url_base}rest/v1/leads/describe.json", headers=self.authenticator.get_auth_header())
+        json_schema = self.configured_json_schema or self.get_json_schema()
+        standard_properties = set(json_schema["properties"])
+        resp = self._session.get(f"{self._url_base}rest/v1/leads/describe.json")
         available_fields = set(x.get("rest").get("name") for x in resp.json().get("result"))
         return list(standard_properties & available_fields)
 
-    def get_json_schema(self) -> Mapping[str, Any]:
-        # TODO: make schema truly dynamic like in stream Activities
-        #  now blocked by https://github.com/airbytehq/airbyte/issues/30530 due to potentially > 500 fields in schema (can cause OOM)
-        return super().get_json_schema()
-
 
 class Activities(MarketoExportBase):
     """

Test Run

1. Connector was built with --use-local-cdk
2. In Configured catalog only leads stream is presented AND 4 properties left

Log output for read command

{"type": "RECORD", "record": {"stream": "leads", "data": {"updatedAt": "2023-01-24T11:10:51Z", "company": "Test Company", "site": null, "id": 885}, "emitted_at": 1718628626222}}
{"type": "RECORD", "record": {"stream": "leads", "data": {"updatedAt": "2023-01-24T11:10:51Z", "company": null, "site": null, "id": 888}, "emitted_at": 1718628626223}}
{"type": "RECORD", "record": {"stream": "leads", "data": {"updatedAt": "2023-01-24T11:10:51Z", "company": null, "site": null, "id": 889}, "emitted_at": 1718628626223}}

Test results

only 4 fields (mentioned in catalog) were extracted

[Alexandre Girard (girarda)]

looks good!