CDK: Remove list endpoint (#29581)

airbyte-cdk/python/airbyte_cdk/connector_builder/README.md

@@ -11,17 +11,17 @@ python main.py read --config path/to/config --catalog path/to/catalog
 ```
 
 Note:
-- Requires the keys `__injected_declarative_manifest` and `__command` in its config, where `__injected_declarative_manifest` is a JSON manifest and `__command` is one of the commands handled by the ConnectorBuilderHandler (`stream_read`, `list_streams`, or `resolve_manifest`), i.e.
+- Requires the keys `__injected_declarative_manifest` and `__command` in its config, where `__injected_declarative_manifest` is a JSON manifest and `__command` is one of the commands handled by the ConnectorBuilderHandler (`stream_read` or `resolve_manifest`), i.e.
 ```
 {
   "config": <normal config>,
   "__injected_declarative_manifest": {...},
-  "__command": <"resolve_manifest" | "list_streams" | "test_read">
+  "__command": <"resolve_manifest" | "test_read">
 }
 ```
 *See [ConnectionSpecification](https://docs.airbyte.com/understanding-airbyte/airbyte-protocol/#actor-specification) for details on the `"config"` key if needed.
 
-- When the `__command` is `list_streams` or `resolve_manifest`, the argument to `catalog` should be an empty string.
+- When the `__command` is `resolve_manifest`, the argument to `catalog` should be an empty string.
 
 ### Locally running the docker image
 

airbyte-cdk/python/airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -4,18 +4,15 @@
 
 import dataclasses
 from datetime import datetime
-from typing import Any, Dict, List, Mapping
-from urllib.parse import urljoin
+from typing import Any, Mapping
 
 from airbyte_cdk.connector_builder.message_grouper import MessageGrouper
 from airbyte_cdk.models import AirbyteMessage, AirbyteRecordMessage, ConfiguredAirbyteCatalog
 from airbyte_cdk.models import Type
 from airbyte_cdk.models import Type as MessageType
 from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
-from airbyte_cdk.sources.declarative.declarative_stream import DeclarativeStream
 from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import ModelToComponentFactory
-from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE = 5
@@ -89,44 +86,5 @@ def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
         return error.as_airbyte_message()
 
 
-def list_streams(source: ManifestDeclarativeSource, config: Dict[str, Any]) -> AirbyteMessage:
-    try:
-        streams = [
-            {
-                "name": http_stream.name,
-                "url": urljoin(
-                    http_stream.requester.get_url_base(),
-                    http_stream.requester.get_path(stream_state=None, stream_slice=None, next_page_token=None),
-                ),
-            }
-            for http_stream in _get_http_streams(source, config)
-        ]
-        return AirbyteMessage(
-            type=Type.RECORD,
-            record=AirbyteRecordMessage(
-                data={"streams": streams},
-                emitted_at=_emitted_at(),
-                stream="list_streams",
-            ),
-        )
-    except Exception as exc:
-        return AirbyteTracedException.from_exception(exc, message=f"Error listing streams: {str(exc)}").as_airbyte_message()
-
-
-def _get_http_streams(source: ManifestDeclarativeSource, config: Dict[str, Any]) -> List[SimpleRetriever]:
-    http_streams = []
-    for stream in source.streams(config=config):
-        if isinstance(stream, DeclarativeStream):
-            if isinstance(stream.retriever, SimpleRetriever):
-                http_streams.append(stream.retriever)
-            else:
-                raise TypeError(
-                    f"A declarative stream should only have a retriever of type SimpleRetriever, but received: {stream.retriever.__class__}"
-                )
-        else:
-            raise TypeError(f"A declarative source should only contain streams of type DeclarativeStream, but received: {stream.__class__}")
-    return http_streams
-
-
 def _emitted_at() -> int:
     return int(datetime.now().timestamp()) * 1000

airbyte-cdk/python/airbyte_cdk/connector_builder/main.py

@@ -7,16 +7,9 @@
 from typing import Any, List, Mapping, Optional, Tuple
 
 from airbyte_cdk.connector import BaseConnector
-from airbyte_cdk.connector_builder.connector_builder_handler import (
-    TestReadLimits,
-    create_source,
-    get_limits,
-    list_streams,
-    read_stream,
-    resolve_manifest,
-)
+from airbyte_cdk.connector_builder.connector_builder_handler import TestReadLimits, create_source, get_limits, read_stream, resolve_manifest
 from airbyte_cdk.entrypoint import AirbyteEntrypoint
-from airbyte_cdk.models import ConfiguredAirbyteCatalog
+from airbyte_cdk.models import AirbyteMessage, ConfiguredAirbyteCatalog
 from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
@@ -50,19 +43,17 @@ def get_config_and_catalog_from_args(args: List[str]) -> Tuple[str, Mapping[str,
 
 def handle_connector_builder_request(
     source: ManifestDeclarativeSource, command: str, config: Mapping[str, Any], catalog: Optional[ConfiguredAirbyteCatalog], limits: TestReadLimits
-):
+) -> AirbyteMessage:
     if command == "resolve_manifest":
         return resolve_manifest(source)
     elif command == "test_read":
         assert catalog is not None, "`test_read` requires a valid `ConfiguredAirbyteCatalog`, got None."
         return read_stream(source, config, catalog, limits)
-    elif command == "list_streams":
-        return list_streams(source, config)
     else:
         raise ValueError(f"Unrecognized command {command}.")
 
 
-def handle_request(args: List[str]):
+def handle_request(args: List[str]) -> AirbyteMessage:
     command, config, catalog = get_config_and_catalog_from_args(args)
     limits = get_limits(config)
     source = create_source(config, limits)

airbyte-cdk/python/unit_tests/connector_builder/test_connector_builder_handler.py

@@ -20,7 +20,6 @@
     TestReadLimits,
     create_source,
     get_limits,
-    list_streams,
     resolve_manifest,
 )
 from airbyte_cdk.connector_builder.main import handle_connector_builder_request, handle_request, read_stream
@@ -43,7 +42,6 @@
 from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from airbyte_cdk.sources.declarative.retrievers import SimpleRetrieverTestReadDecorator
 from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
-from airbyte_cdk.sources.streams.core import Stream
 from unit_tests.connector_builder.utils import create_configured_catalog
 
 _stream_name = "stream_with_custom_requester"
@@ -537,7 +535,7 @@ def check_config_against_spec(self):
 )
 def test_invalid_protocol_command(command, valid_resolve_manifest_config_file):
     config = copy.deepcopy(RESOLVE_MANIFEST_CONFIG)
-    config["__command"] = "list_streams"
+    config["__command"] = "resolve_manifest"
     with pytest.raises(SystemExit):
         handle_request([command, "--config", str(valid_resolve_manifest_config_file), "--catalog", ""])
 
@@ -567,71 +565,6 @@ def manifest_declarative_source():
     return mock.Mock(spec=ManifestDeclarativeSource, autospec=True)
 
 
-def test_list_streams(manifest_declarative_source):
-    manifest_declarative_source.streams.return_value = [
-        create_mock_declarative_stream(create_mock_retriever("a name", "https://a-url-base.com", "a-path")),
-        create_mock_declarative_stream(create_mock_retriever("another name", "https://another-url-base.com", "another-path")),
-    ]
-
-    result = list_streams(manifest_declarative_source, {})
-
-    assert result.type == MessageType.RECORD
-    assert result.record.stream == "list_streams"
-    assert result.record.data == {
-        "streams": [
-            {"name": "a name", "url": "https://a-url-base.com/a-path"},
-            {"name": "another name", "url": "https://another-url-base.com/another-path"},
-        ]
-    }
-
-
-def test_given_stream_is_not_declarative_stream_when_list_streams_then_return_exception_message(manifest_declarative_source):
-    manifest_declarative_source.streams.return_value = [mock.Mock(spec=Stream)]
-
-    error_message = list_streams(manifest_declarative_source, {})
-
-    assert error_message.type == MessageType.TRACE
-    assert error_message.trace.error.message.startswith("Error listing streams")
-    assert "A declarative source should only contain streams of type DeclarativeStream" in error_message.trace.error.internal_message
-
-
-def test_given_declarative_stream_retriever_is_not_http_when_list_streams_then_return_exception_message(manifest_declarative_source):
-    declarative_stream = mock.Mock(spec=DeclarativeStream)
-    # `spec=DeclarativeStream` is needed for `isinstance` work but `spec` does not expose dataclasses fields, so we create one ourselves
-    declarative_stream.retriever = mock.Mock()
-    manifest_declarative_source.streams.return_value = [declarative_stream]
-
-    error_message = list_streams(manifest_declarative_source, {})
-
-    assert error_message.type == MessageType.TRACE
-    assert error_message.trace.error.message.startswith("Error listing streams")
-    assert "A declarative stream should only have a retriever of type SimpleRetriever" in error_message.trace.error.internal_message
-
-
-def test_given_unexpected_error_when_list_streams_then_return_exception_message(manifest_declarative_source):
-    manifest_declarative_source.streams.side_effect = Exception("unexpected error")
-
-    error_message = list_streams(manifest_declarative_source, {})
-
-    assert error_message.type == MessageType.TRACE
-    assert error_message.trace.error.message.startswith("Error listing streams")
-    assert "unexpected error" == error_message.trace.error.internal_message
-
-
-def test_list_streams_integration_test():
-    config = copy.deepcopy(RESOLVE_MANIFEST_CONFIG)
-    command = "list_streams"
-    config["__command"] = command
-    source = ManifestDeclarativeSource(MANIFEST)
-    limits = TestReadLimits()
-
-    list_streams = handle_connector_builder_request(source, command, config, None, limits)
-
-    assert list_streams.record.data == {
-        "streams": [{"name": "stream_with_custom_requester", "url": "https://api.sendgrid.com/v3/marketing/lists"}]
-    }
-
-
 def create_mock_retriever(name, url_base, path):
     http_stream = mock.Mock(spec=SimpleRetriever, autospec=True)
     http_stream.name = name