Revise stream_count semantics: None=check all, 0=check none

Co-Authored-By: gl_anatolii.yatsuk <gl_anatolii.yatsuk@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/checks/check_stream.py

@@ -36,7 +36,7 @@ class DynamicStreamCheckConfig:
     and type enforcement."""
 
     dynamic_stream_name: str
-    stream_count: int = 0
+    stream_count: Optional[int] = None
 
 
 @dataclass
@@ -162,13 +162,16 @@ def _check_generated_streams_availability(
         generated_streams: List[Dict[str, Any]],
         stream_name_to_stream: Dict[str, Union[Stream, AbstractStream]],
         logger: logging.Logger,
-        max_count: int,
+        max_count: Optional[int],
     ) -> Tuple[bool, Any]:
         """Checks availability of generated dynamic streams.
 
-        If max_count is 0 or negative, all generated streams are checked.
+        If `max_count` is `None`, all generated streams are checked. If `max_count` is 0,
+        no streams are checked. Otherwise, the first `max_count` streams are checked.
         """
-        streams_to_check = generated_streams if max_count <= 0 else generated_streams[:max_count]
+        streams_to_check = (
+            generated_streams if max_count is None else generated_streams[:max_count]
+        )
         for declarative_stream in streams_to_check:
             stream = stream_name_to_stream[declarative_stream["name"]]
             try:

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -358,9 +358,9 @@ definitions:
         type: string
       stream_count:
         title: Stream Count
-        description: The number of streams to attempt reading from during a check operation. If `stream_count` exceeds the total number of available streams, the minimum of the two values will be used.
+        description: The number of streams to attempt reading from during a check operation. If unset, all generated streams are checked. If set to 0, no streams are checked. If set to a positive value greater than the total number of available streams, all streams are checked.
         type: integer
-        default: 0
+        minimum: 0
   CheckDynamicStream:
     title: Dynamic Streams to Check
     description: (This component is experimental. Use at your own risk.) Defines the dynamic streams to try reading when running a check operation.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -58,8 +58,9 @@ class DynamicStreamCheckConfig(BaseModel):
         ..., description="The dynamic stream name.", title="Dynamic Stream Name"
     )
     stream_count: Optional[int] = Field(
-        0,
-        description="The number of streams to attempt reading from during a check operation. If `stream_count` exceeds the total number of available streams, the minimum of the two values will be used.",
+        None,
+        description="The number of streams to attempt reading from during a check operation. If unset, all generated streams are checked. If set to 0, no streams are checked. If set to a positive value greater than the total number of available streams, all streams are checked.",
+        ge=0,
         title="Stream Count",
     )
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1246,7 +1246,7 @@ def create_dynamic_stream_check_config(
     ) -> DynamicStreamCheckConfig:
         return DynamicStreamCheckConfig(
             dynamic_stream_name=model.dynamic_stream_name,
-            stream_count=model.stream_count or 0,
+            stream_count=model.stream_count,
         )
 
     def create_check_stream(

unit_tests/sources/declarative/checks/test_check_stream.py

@@ -504,8 +504,8 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
             False,
             200,
             [],
-            1,
-            id="test_stream_count_zero_checks_all_streams",
+            0,
+            id="test_stream_count_zero_checks_no_streams",
         ),
         pytest.param(
             {
@@ -520,12 +520,50 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
                     ],
                 }
             },
+            Status.SUCCEEDED,
+            False,
+            404,
+            ["Not found. The requested resource was not found on the server."],
+            0,
+            id="test_stream_count_zero_skips_failing_streams",
+        ),
+        pytest.param(
+            {
+                "check": {
+                    "type": "CheckStream",
+                    "dynamic_streams_check_configs": [
+                        {
+                            "type": "DynamicStreamCheckConfig",
+                            "dynamic_stream_name": "http_dynamic_stream",
+                        },
+                    ],
+                }
+            },
+            Status.SUCCEEDED,
+            False,
+            200,
+            [],
+            1,
+            id="test_stream_count_unset_checks_all_streams",
+        ),
+        pytest.param(
+            {
+                "check": {
+                    "type": "CheckStream",
+                    "dynamic_streams_check_configs": [
+                        {
+                            "type": "DynamicStreamCheckConfig",
+                            "dynamic_stream_name": "http_dynamic_stream",
+                        },
+                    ],
+                }
+            },
             Status.FAILED,
             False,
             404,
             ["Not found. The requested resource was not found on the server."],
             0,
-            id="test_stream_count_zero_failed",
+            id="test_stream_count_unset_failed",
         ),
         pytest.param(
             {"check": {"type": "CheckStream", "stream_names": ["non_existent_stream"]}},
@@ -722,6 +760,32 @@ def test_check_stream_missing_fields():
         )
 
 
+def test_check_stream_negative_stream_count():
+    """Test that a ValidationError is raised when stream_count is negative."""
+    manifest = {
+        **deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT),
+        **{
+            "check": {
+                "type": "CheckStream",
+                "dynamic_streams_check_configs": [
+                    {
+                        "type": "DynamicStreamCheckConfig",
+                        "dynamic_stream_name": "http_dynamic_stream",
+                        "stream_count": -1,
+                    }
+                ],
+            }
+        },
+    }
+    with pytest.raises(ValidationError):
+        ConcurrentDeclarativeSource(
+            source_config=manifest,
+            config=_CONFIG,
+            catalog=None,
+            state=None,
+        )
+
+
 def test_check_stream_only_type_provided():
     manifest = {**deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT), **{"check": {"type": "CheckStream"}}}
     source = ConcurrentDeclarativeSource(