Merge branch 'main' into devin/1763137629-excel-parser-openpyxl-fallback

Author: agarctfi

.github/workflows/label-community-prs.yml

@@ -0,0 +1,28 @@
+name: Label Community PRs
+
+# This workflow automatically adds the "community" label to PRs from forks.
+# This enables automatic tracking on the Community PRs project board.
+
+on:
+  pull_request_target:
+    types:
+      - opened
+      - reopened
+
+jobs:
+  label-community-pr:
+    name: Add "Community" Label to PR
+    # Only run for PRs from forks
+    if: github.event.pull_request.head.repo.fork == true
+    runs-on: ubuntu-24.04
+    permissions:
+      issues: write
+      pull-requests: write
+    steps:
+      - name: Add community label
+        # This action uses GitHub's addLabels API, which is idempotent.
+        # If the label already exists, the API call succeeds without error.
+        uses: actions-ecosystem/action-add-labels@bd52874380e3909a1ac983768df6976535ece7f8 # v1.1.3
+        with:
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          labels: community

.github/workflows/release_drafter.yml

@@ -11,6 +11,10 @@ on:
   # pull_request_target:
   #   types: [opened, reopened, synchronize]
 
+concurrency:
+  group: release-drafter-${{ github.ref }}
+  cancel-in-progress: false
+
 jobs:
   update_release_draft:
     permissions:
@@ -19,7 +23,8 @@ jobs:
     runs-on: ubuntu-24.04
     steps:
       # Drafts the next Release notes as Pull Requests are merged into "main"
-      - uses: release-drafter/release-drafter@v6
+      # Pinned to v6.0.0 to avoid v6.1.0 bug: https://github.com/release-drafter/release-drafter/issues/1425
+      - uses: release-drafter/release-drafter@v6.0.0
         with:
           config-name: release-drafter.yml
         env:

airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py

@@ -341,29 +341,41 @@ def streams(self, config: Mapping[str, Any]) -> List[Union[Stream, AbstractStrea
     def _initialize_cache_for_parent_streams(
         stream_configs: List[Dict[str, Any]],
     ) -> List[Dict[str, Any]]:
+        """Enable caching for parent streams unless explicitly disabled.
+
+        Caching is enabled by default for parent streams to optimize performance when the same
+        parent data is needed by multiple child streams. However, explicit `use_cache: false`
+        settings are respected for streams that cannot use caching (e.g., scroll-based pagination
+        APIs where caching causes duplicate records).
+        """
         parent_streams = set()
 
+        def _set_cache_if_not_disabled(requester: Dict[str, Any]) -> None:
+            """Set use_cache to True only if not explicitly disabled."""
+            if requester.get("use_cache") is not False:
+                requester["use_cache"] = True
+
         def update_with_cache_parent_configs(
             parent_configs: list[dict[str, Any]],
         ) -> None:
             for parent_config in parent_configs:
                 parent_streams.add(parent_config["stream"]["name"])
                 if parent_config["stream"]["type"] == "StateDelegatingStream":
-                    parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"][
-                        "use_cache"
-                    ] = True
-                    parent_config["stream"]["incremental_stream"]["retriever"]["requester"][
-                        "use_cache"
-                    ] = True
+                    _set_cache_if_not_disabled(
+                        parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"]
+                    )
+                    _set_cache_if_not_disabled(
+                        parent_config["stream"]["incremental_stream"]["retriever"]["requester"]
+                    )
                 else:
-                    parent_config["stream"]["retriever"]["requester"]["use_cache"] = True
+                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"]["requester"])
 
         for stream_config in stream_configs:
             if stream_config.get("incremental_sync", {}).get("parent_stream"):
                 parent_streams.add(stream_config["incremental_sync"]["parent_stream"]["name"])
-                stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"][
-                    "use_cache"
-                ] = True
+                _set_cache_if_not_disabled(
+                    stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"]
+                )
 
             elif stream_config.get("retriever", {}).get("partition_router", {}):
                 partition_router = stream_config["retriever"]["partition_router"]
@@ -380,14 +392,14 @@ def update_with_cache_parent_configs(
         for stream_config in stream_configs:
             if stream_config["name"] in parent_streams:
                 if stream_config["type"] == "StateDelegatingStream":
-                    stream_config["full_refresh_stream"]["retriever"]["requester"]["use_cache"] = (
-                        True
+                    _set_cache_if_not_disabled(
+                        stream_config["full_refresh_stream"]["retriever"]["requester"]
                     )
-                    stream_config["incremental_stream"]["retriever"]["requester"]["use_cache"] = (
-                        True
+                    _set_cache_if_not_disabled(
+                        stream_config["incremental_stream"]["retriever"]["requester"]
                     )
                 else:
-                    stream_config["retriever"]["requester"]["use_cache"] = True
+                    _set_cache_if_not_disabled(stream_config["retriever"]["requester"])
         return stream_configs
 
     def spec(self, logger: logging.Logger) -> ConnectorSpecification:

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -424,29 +424,41 @@ def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]:  # type: i
     def _initialize_cache_for_parent_streams(
         stream_configs: List[Dict[str, Any]],
     ) -> List[Dict[str, Any]]:
+        """Enable caching for parent streams unless explicitly disabled.
+
+        Caching is enabled by default for parent streams to optimize performance when the same
+        parent data is needed by multiple child streams. However, explicit `use_cache: false`
+        settings are respected for streams that cannot use caching (e.g., scroll-based pagination
+        APIs where caching causes duplicate records).
+        """
         parent_streams = set()
 
+        def _set_cache_if_not_disabled(requester: Dict[str, Any]) -> None:
+            """Set use_cache to True only if not explicitly disabled."""
+            if requester.get("use_cache") is not False:
+                requester["use_cache"] = True
+
         def update_with_cache_parent_configs(
             parent_configs: list[dict[str, Any]],
         ) -> None:
             for parent_config in parent_configs:
                 parent_streams.add(parent_config["stream"]["name"])
                 if parent_config["stream"]["type"] == "StateDelegatingStream":
-                    parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"][
-                        "use_cache"
-                    ] = True
-                    parent_config["stream"]["incremental_stream"]["retriever"]["requester"][
-                        "use_cache"
-                    ] = True
+                    _set_cache_if_not_disabled(
+                        parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"]
+                    )
+                    _set_cache_if_not_disabled(
+                        parent_config["stream"]["incremental_stream"]["retriever"]["requester"]
+                    )
                 else:
-                    parent_config["stream"]["retriever"]["requester"]["use_cache"] = True
+                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"]["requester"])
 
         for stream_config in stream_configs:
             if stream_config.get("incremental_sync", {}).get("parent_stream"):
                 parent_streams.add(stream_config["incremental_sync"]["parent_stream"]["name"])
-                stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"][
-                    "use_cache"
-                ] = True
+                _set_cache_if_not_disabled(
+                    stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"]
+                )
 
             elif stream_config.get("retriever", {}).get("partition_router", {}):
                 partition_router = stream_config["retriever"]["partition_router"]
@@ -463,14 +475,14 @@ def update_with_cache_parent_configs(
         for stream_config in stream_configs:
             if stream_config["name"] in parent_streams:
                 if stream_config["type"] == "StateDelegatingStream":
-                    stream_config["full_refresh_stream"]["retriever"]["requester"]["use_cache"] = (
-                        True
+                    _set_cache_if_not_disabled(
+                        stream_config["full_refresh_stream"]["retriever"]["requester"]
                     )
-                    stream_config["incremental_stream"]["retriever"]["requester"]["use_cache"] = (
-                        True
+                    _set_cache_if_not_disabled(
+                        stream_config["incremental_stream"]["retriever"]["requester"]
                     )
                 else:
-                    stream_config["retriever"]["requester"]["use_cache"] = True
+                    _set_cache_if_not_disabled(stream_config["retriever"]["requester"])
         return stream_configs
 
     def spec(self, logger: logging.Logger) -> ConnectorSpecification:

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2286,7 +2286,7 @@ definitions:
           - "$ref": "#/definitions/CustomErrorHandler"
       use_cache:
         title: Use Cache
-        description: Enables stream requests caching. This field is automatically set by the CDK.
+        description: Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).
         type: boolean
         default: false
       $parameters:

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -4339,8 +4339,14 @@ def _get_catalog_defined_cursor_field(
         configured_stream = self._stream_name_to_configured_stream.get(stream_name)
 
         # Depending on the operation is being performed, there may not be a configured stream yet. In this
-        # case we return None which will then use the default cursor field defined on the cursor model
-        if not configured_stream or not configured_stream.cursor_field:
+        # case we return None which will then use the default cursor field defined on the cursor model.
+        # We also treat cursor_field: [""] (list with empty string) as no cursor field, since this can
+        # occur when the platform serializes "no cursor configured" streams incorrectly.
+        if (
+            not configured_stream
+            or not configured_stream.cursor_field
+            or not configured_stream.cursor_field[0]
+        ):
             return None
         elif len(configured_stream.cursor_field) > 1:
             raise ValueError(

airbyte_cdk/sources/streams/concurrent/helpers.py

@@ -34,7 +34,8 @@ def get_cursor_field_from_stream(stream: Stream) -> Optional[str]:
             raise ValueError(
                 f"Nested cursor fields are not supported. Got {stream.cursor_field} for {stream.name}"
             )
-        elif len(stream.cursor_field) == 0:
+        elif len(stream.cursor_field) == 0 or not stream.cursor_field[0]:
+            # Treat cursor_field: [""] (list with empty string) as no cursor field
             return None
         else:
             return stream.cursor_field[0]

unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py

@@ -1652,6 +1652,112 @@ def test_only_parent_streams_use_cache():
     assert not get_retriever(streams[2]).requester.use_cache
 
 
+def test_parent_stream_respects_explicit_use_cache_false():
+    """Test that explicit use_cache: false is respected for parent streams.
+
+    This is important for APIs that use scroll-based pagination (like Intercom's /companies/scroll
+    endpoint), where caching must be disabled because the same scroll_param is returned in
+    pagination responses, causing duplicate records and infinite pagination loops.
+    """
+    # Parent stream with explicit use_cache: false
+    companies_stream = {
+        "type": "DeclarativeStream",
+        "$parameters": {
+            "name": "companies",
+            "primary_key": "id",
+            "url_base": "https://api.intercom.io/",
+        },
+        "schema_loader": {
+            "name": "{{ parameters.stream_name }}",
+            "file_path": "./source_intercom/schemas/{{ parameters.name }}.yaml",
+        },
+        "retriever": {
+            "paginator": {
+                "type": "DefaultPaginator",
+                "page_token_option": {"type": "RequestPath"},
+                "pagination_strategy": {
+                    "type": "CursorPagination",
+                    "cursor_value": "{{ response.get('scroll_param') }}",
+                    "page_size": 100,
+                },
+            },
+            "requester": {
+                "path": "companies/scroll",
+                "use_cache": False,  # Explicitly disabled for scroll-based pagination
+                "authenticator": {
+                    "type": "BearerAuthenticator",
+                    "api_token": "{{ config['api_key'] }}",
+                },
+            },
+            "record_selector": {"extractor": {"type": "DpathExtractor", "field_path": ["data"]}},
+        },
+    }
+
+    manifest = {
+        "version": "0.29.3",
+        "definitions": {},
+        "streams": [
+            deepcopy(companies_stream),
+            {
+                "type": "DeclarativeStream",
+                "$parameters": {
+                    "name": "company_segments",
+                    "primary_key": "id",
+                    "url_base": "https://api.intercom.io/",
+                },
+                "schema_loader": {
+                    "name": "{{ parameters.stream_name }}",
+                    "file_path": "./source_intercom/schemas/{{ parameters.name }}.yaml",
+                },
+                "retriever": {
+                    "paginator": {"type": "NoPagination"},
+                    "requester": {
+                        "path": "companies/{{ stream_partition.parent_id }}/segments",
+                        "authenticator": {
+                            "type": "BearerAuthenticator",
+                            "api_token": "{{ config['api_key'] }}",
+                        },
+                    },
+                    "record_selector": {
+                        "extractor": {"type": "DpathExtractor", "field_path": ["data"]}
+                    },
+                    "partition_router": {
+                        "parent_stream_configs": [
+                            {
+                                "parent_key": "id",
+                                "partition_field": "parent_id",
+                                "stream": deepcopy(companies_stream),
+                            }
+                        ],
+                        "type": "SubstreamPartitionRouter",
+                    },
+                },
+            },
+        ],
+        "check": {"type": "CheckStream", "stream_names": ["companies"]},
+    }
+    source = ManifestDeclarativeSource(source_config=manifest)
+
+    streams = source.streams({})
+    assert len(streams) == 2
+
+    # Main stream with explicit use_cache: false should remain false (parent for substream)
+    assert streams[0].name == "companies"
+    # use_cache should remain False because it was explicitly set to False
+    assert not get_retriever(streams[0]).requester.use_cache
+
+    # Substream
+    assert streams[1].name == "company_segments"
+
+    # Parent stream created for substream should also respect use_cache: false
+    stream_slicer = streams[1]._stream_partition_generator._stream_slicer
+    assert stream_slicer.parent_stream_configs[0].stream.name == "companies"
+    # The parent stream in the substream config should also have use_cache: false
+    assert not stream_slicer.parent_stream_configs[
+        0
+    ].stream._stream_partition_generator._partition_factory._retriever.requester.use_cache
+
+
 def _run_read(manifest: Mapping[str, Any], stream_name: str) -> List[AirbyteMessage]:
     catalog = ConfiguredAirbyteCatalog(
         streams=[

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -5051,6 +5051,12 @@ def test_create_stream_with_multiple_schema_loaders():
             "updated_at",
             id="test_allow_catalog_defined_cursor_field_false_defaults_to_stream_defined_cursor_field",
         ),
+        pytest.param(
+            True,
+            "",
+            "updated_at",
+            id="test_empty_string_catalog_cursor_field_defaults_to_stream_defined_cursor_field",
+        ),
     ],
 )
 def test_catalog_defined_cursor_field(