add ConditionalStreams component to low-code framework

Author: brianjlai

.pre-commit-config.yaml

@@ -74,3 +74,4 @@ repos:
         args: [--config-file=mypy.ini, --show-column-numbers]
         files: ^airbyte_cdk/
         pass_filenames: true
+        additional_dependencies: ["types-requests", "types-PyYAML"]

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -25,6 +25,7 @@ properties:
     type: array
     items:
       anyOf:
+        - "$ref": "#/definitions/ConditionalStreams"
         - "$ref": "#/definitions/DeclarativeStream"
         - "$ref": "#/definitions/StateDelegatingStream"
   dynamic_streams:
@@ -424,6 +425,36 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  ConditionalStreams:
+    title: Conditional Streams
+    description: Streams that are only available while performing a connector operation when the condition is met.
+    type: object
+    required:
+      - type
+      - streams
+      - condition
+    properties:
+      type:
+        type: string
+        enum: [ConditionalStreams]
+      condition:
+        title: Condition
+        description: Condition that will be evaluated to determine if a set of streams should be available.
+        type: string
+        interpolation_context:
+          - config
+          - parameters
+        examples:
+          - "{{ config['is_sandbox'] }}"
+      streams:
+        title: Streams
+        description: Streams that will be used during an operation based on the condition.
+        type: array
+        items:
+          "$ref": "#/definitions/DeclarativeStream"
+      $parameters:
+        type: object
+        additionalProperties: true
   ConstantBackoffStrategy:
     title: Constant Backoff
     description: Backoff strategy with a constant backoff interval.

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -8,7 +8,7 @@
 from copy import deepcopy
 from importlib import metadata
 from types import ModuleType
-from typing import Any, Dict, Iterator, List, Mapping, MutableMapping, Optional, Set
+from typing import Any, Dict, Iterator, List, Mapping, Optional, Set
 
 import orjson
 import yaml
@@ -35,6 +35,9 @@
 from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConditionalStreams as ConditionalStreamsModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DeclarativeStream as DeclarativeStreamModel,
 )
@@ -306,20 +309,37 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         if api_budget_model:
             self._constructor.set_api_budget(api_budget_model, config)
 
-        source_streams = [
-            self._constructor.create_component(
-                (
-                    StateDelegatingStreamModel
-                    if stream_config.get("type") == StateDelegatingStreamModel.__name__
-                    else DeclarativeStreamModel
-                ),
-                stream_config,
-                config,
-                emit_connector_builder_messages=self._emit_connector_builder_messages,
-            )
-            for stream_config in self._initialize_cache_for_parent_streams(deepcopy(stream_configs))
-        ]
-
+        source_streams = []
+        for stream_config in self._initialize_cache_for_parent_streams(deepcopy(stream_configs)):
+            match stream_config.get("type"):
+                case StateDelegatingStreamModel.__name__:
+                    source_streams.append(
+                        self._constructor.create_component(
+                            StateDelegatingStreamModel,
+                            stream_config,
+                            config,
+                            emit_connector_builder_messages=self._emit_connector_builder_messages,
+                        )
+                    )
+                case ConditionalStreamsModel.__name__:
+                    # ConditionalStreamsModel returns a list of DeclarativeStreams so it must be extended
+                    source_streams.extend(
+                        self._constructor.create_component(
+                            ConditionalStreamsModel,
+                            stream_config,
+                            config,
+                            emit_connector_builder_messages=self._emit_connector_builder_messages,
+                        )
+                    )
+                case DeclarativeStreamModel.__name__:
+                    source_streams.append(
+                        self._constructor.create_component(
+                            DeclarativeStreamModel,
+                            stream_config,
+                            config,
+                            emit_connector_builder_messages=self._emit_connector_builder_messages,
+                        )
+                    )
         return source_streams
 
     @staticmethod
@@ -363,17 +383,26 @@ def update_with_cache_parent_configs(
                             update_with_cache_parent_configs(router["parent_stream_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
-                    )
-                    stream_config["incremental_stream"]["retriever"]["requester"]["use_cache"] = (
-                        True
-                    )
+            current_stream_configs = []
+            if stream_config.get("type") == ConditionalStreamsModel.__name__:
+                current_stream_configs = [
+                    conditional_stream_config
+                    for conditional_stream_config in stream_config.get("streams") or []
+                    if conditional_stream_config["name"] in parent_streams
+                ]
+            elif stream_config["name"] in parent_streams:
+                current_stream_configs = [stream_config]
+
+            for current_stream_config in current_stream_configs:
+                if current_stream_config["type"] == "StateDelegatingStream":
+                    current_stream_config["full_refresh_stream"]["retriever"]["requester"][
+                        "use_cache"
+                    ] = True
+                    current_stream_config["incremental_stream"]["retriever"]["requester"][
+                        "use_cache"
+                    ] = True
                 else:
-                    stream_config["retriever"]["requester"]["use_cache"] = True
-
+                    current_stream_config["retriever"]["requester"]["use_cache"] = True
         return stream_configs
 
     def spec(self, logger: logging.Logger) -> ConnectorSpecification:

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -2168,7 +2170,7 @@ class Config:
 
     type: Literal["DeclarativeSource"]
     check: Union[CheckStream, CheckDynamicStream]
-    streams: List[Union[DeclarativeStream, StateDelegatingStream]]
+    streams: List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]]
     dynamic_streams: Optional[List[DynamicDeclarativeStream]] = None
     version: str = Field(
         ...,
@@ -2201,7 +2203,9 @@ class Config:
 
     type: Literal["DeclarativeSource"]
     check: Union[CheckStream, CheckDynamicStream]
-    streams: Optional[List[Union[DeclarativeStream, StateDelegatingStream]]] = None
+    streams: Optional[List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]]] = (
+        None
+    )
     dynamic_streams: List[DynamicDeclarativeStream]
     version: str = Field(
         ...,
@@ -2280,6 +2284,22 @@ class Config:
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class ConditionalStreams(BaseModel):
+    type: Literal["ConditionalStreams"]
+    condition: str = Field(
+        ...,
+        description="Condition that will be evaluated to determine if a set of streams should be available.",
+        examples=["{{ config['is_sandbox'] }}"],
+        title="Condition",
+    )
+    streams: List[DeclarativeStream] = Field(
+        ...,
+        description="Streams that will be used during an operation based on the condition.",
+        title="Streams",
+    )
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
+
+
 class FileUploader(BaseModel):
     type: Literal["FileUploader"]
     requester: Union[HttpRequester, CustomRequester] = Field(
@@ -2936,6 +2956,7 @@ class DynamicDeclarativeStream(BaseModel):
 DeclarativeSource1.update_forward_refs()
 DeclarativeSource2.update_forward_refs()
 SelectiveAuthenticator.update_forward_refs()
+ConditionalStreams.update_forward_refs()
 FileUploader.update_forward_refs()
 DeclarativeStream.update_forward_refs()
 SessionTokenAuthenticator.update_forward_refs()

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -105,7 +105,7 @@
     PerPartitionWithGlobalCursor,
     ResumableFullRefreshCursor,
 )
-from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
+from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean, InterpolatedString
 from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
 from airbyte_cdk.sources.declarative.migrations.legacy_to_per_partition_state_migration import (
     LegacyToPerPartitionStateMigration,
@@ -156,6 +156,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConcurrencyLevel as ConcurrencyLevelModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConditionalStreams as ConditionalStreamsModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConfigAddFields as ConfigAddFieldsModel,
 )
@@ -656,6 +659,7 @@ def _init_mappings(self) -> None:
             CheckDynamicStreamModel: self.create_check_dynamic_stream,
             CompositeErrorHandlerModel: self.create_composite_error_handler,
             ConcurrencyLevelModel: self.create_concurrency_level,
+            ConditionalStreamsModel: self.create_conditional_streams,
             ConfigMigrationModel: self.create_config_migration,
             ConfigAddFieldsModel: self.create_config_add_fields,
             ConfigRemapFieldModel: self.create_config_remap_field,
@@ -1619,6 +1623,22 @@ def create_concurrent_cursor_from_perpartition_cursor(
             use_global_cursor=use_global_cursor,
         )
 
+    def create_conditional_streams(
+        self, model: ConditionalStreamsModel, config: Config, **kwargs: Any
+    ) -> List[DeclarativeStream]:
+        condition = InterpolatedBoolean(
+            condition=model.condition, parameters=model.parameters or {}
+        )
+        should_include_streams = condition.eval(config=config)
+        return (
+            [
+                self._create_component_from_model(stream, config=config, **kwargs)
+                for stream in model.streams
+            ]
+            if should_include_streams
+            else []
+        )
+
     @staticmethod
     def create_constant_backoff_strategy(
         model: ConstantBackoffStrategyModel, config: Config, **kwargs: Any
@@ -3150,12 +3170,12 @@ def _get_url() -> str:
             This is needed because the URL is not set until the requester is created.
             """
 
-            _url = (
+            _url: str = (
                 model.requester.url
                 if hasattr(model.requester, "url") and model.requester.url is not None
                 else requester.get_url()
             )
-            _url_base = (
+            _url_base: str = (
                 model.requester.url_base
                 if hasattr(model.requester, "url_base") and model.requester.url_base is not None
                 else requester.get_url_base()

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -59,6 +59,7 @@
     CompositeErrorHandler as CompositeErrorHandlerModel,
 )
 from airbyte_cdk.sources.declarative.models import ConcurrencyLevel as ConcurrencyLevelModel
+from airbyte_cdk.sources.declarative.models import ConditionalStreams as ConditionalStreamsModel
 from airbyte_cdk.sources.declarative.models import CustomErrorHandler as CustomErrorHandlerModel
 from airbyte_cdk.sources.declarative.models import (
     CustomPartitionRouter as CustomPartitionRouterModel,
@@ -4801,3 +4802,109 @@ def test_create_stream_with_multiple_schema_loaders():
     assert len(schema_loader.schema_loaders) == 2
     assert isinstance(schema_loader.schema_loaders[0], InlineSchemaLoader)
     assert isinstance(schema_loader.schema_loaders[1], InlineSchemaLoader)
+
+
+def test_conditional_streams():
+    content = """
+    lists_stream:
+      type: DeclarativeStream
+      name: lists
+      primary_key: id
+      retriever:
+        paginator:
+        type: "DefaultPaginator"
+        page_size_option:
+          type: RequestOption
+          inject_into: request_parameter
+          field_name: page_size
+        page_token_option:
+          type: RequestPath
+        pagination_strategy:
+          type: "CursorPagination"
+          cursor_value: "{{ response._metadata.next }}"
+          page_size: 10
+        requester:
+          url_base: "https://testing.com"
+          path: "/api/v1/lists"
+          authenticator:
+            type: "BearerAuthenticator"
+            api_token: "{{ config.apikey }}"
+          request_parameters:
+            page_size: 10
+          record_selector:
+          extractor:
+            field_path: ["result"]
+      schema_loader:
+        type: JsonFileSchemaLoader
+        file_path: "./source_test/schemas/lists.yaml"
+    conditions_stream:
+      type: DeclarativeStream
+      name: sometimes
+      primary_key: id
+      retriever:
+        paginator:
+          type: "DefaultPaginator"
+          page_size_option:
+            type: RequestOption
+            inject_into: request_parameter
+            field_name: page_size
+          page_token_option:
+            type: RequestPath
+          pagination_strategy:
+            type: "CursorPagination"
+            cursor_value: "{{ response._metadata.next }}"
+            page_size: 10
+        requester:
+          type: HttpRequester
+          url_base: "https://testing.com"
+          path: "/api/v1/sometimes"
+          authenticator:
+            type: "BearerAuthenticator"
+            api_token: "{{ config.apikey }}"
+          request_parameters:
+            page_size: 10
+        record_selector:
+          type: RecordSelector
+          extractor:
+            type: DpathExtractor
+            field_path: ["result"]
+      schema_loader:
+        type: JsonFileSchemaLoader
+        file_path: "./source_test/schemas/sometimes.yaml"
+    streams:
+      - "#/lists_stream"
+      - type: ConditionalStreams
+        condition: "{{ config['is_sandbox'] }}"
+        streams:
+          - "#/conditions_stream"
+    """
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    resolved_manifest["type"] = "DeclarativeSource"
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["conditions_stream"], {}
+    )
+
+    conditional_streams_manifest = resolved_manifest["streams"][1]
+    conditional_streams_manifest["streams"] = [stream_manifest]
+
+    sandbox_config = {
+        **input_config,
+        "is_sandbox": True,
+    }
+
+    streams = factory.create_component(
+        model_type=ConditionalStreamsModel,
+        component_definition=conditional_streams_manifest,
+        config=sandbox_config,
+    )
+
+    assert len(streams) == 1
+    conditional_stream = streams[0]
+    assert isinstance(conditional_stream, DeclarativeStream)
+    assert conditional_stream.name == "sometimes"
+    assert conditional_stream.primary_key == "id"
+
+    assert isinstance(conditional_stream.retriever, SimpleRetriever)
+    assert conditional_stream.retriever.name == "sometimes"
+    assert conditional_stream.retriever.primary_key == "id"