feat: Add config-free discover for declarative sources with static catalogs

airbyte_cdk/cli/source_declarative_manifest/_run.py

@@ -37,6 +37,7 @@
     AirbyteTraceMessage,
     ConfiguredAirbyteCatalog,
     ConnectorSpecificationSerializer,
+    FailureType,
     TraceType,
     Type,
 )
@@ -108,6 +109,9 @@ def _get_local_yaml_source(args: list[str]) -> SourceLocalYaml:
             config_path=parsed_args.config if hasattr(parsed_args, "config") else None,
         )
     except Exception as error:
+        is_config_free_discover = args[:1] == ["discover"] and not any(
+            arg.startswith("--config") for arg in args
+        )
         print(
             orjson.dumps(
                 AirbyteMessageSerializer.dump(
@@ -117,8 +121,15 @@ def _get_local_yaml_source(args: list[str]) -> SourceLocalYaml:
                             type=TraceType.ERROR,
                             emitted_at=ab_datetime_now().to_epoch_millis(),
                             error=AirbyteErrorTraceMessage(
-                                message=f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}",
+                                message=(
+                                    f"This connector could not run `discover` without a config. Provide one with `--config`. Error: {error}"
+                                    if is_config_free_discover
+                                    else f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}"
+                                ),
                                 stack_trace=traceback.format_exc(),
+                                failure_type=FailureType.config_error
+                                if is_config_free_discover
+                                else None,
                             ),
                         ),
                     )
@@ -222,6 +233,9 @@ def create_declarative_source(
             custom_components_trusted=False,
         )
     except Exception as error:
+        is_config_free_discover = args[:1] == ["discover"] and not any(
+            arg.startswith("--config") for arg in args
+        )
         print(
             orjson.dumps(
                 AirbyteMessageSerializer.dump(
@@ -231,8 +245,15 @@ def create_declarative_source(
                             type=TraceType.ERROR,
                             emitted_at=ab_datetime_now().to_epoch_millis(),
                             error=AirbyteErrorTraceMessage(
-                                message=f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}",
+                                message=(
+                                    f"This connector could not run `discover` without a config. Provide one with `--config`. Error: {error}"
+                                    if is_config_free_discover
+                                    else f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}"
+                                ),
                                 stack_trace=traceback.format_exc(),
+                                failure_type=FailureType.config_error
+                                if is_config_free_discover
+                                else None,
                             ),
                         ),
                     )
@@ -251,8 +272,8 @@ def _parse_inputs_into_config_catalog_state(
 ]:
     config = (
         ConcurrentDeclarativeSource.read_config(parsed_args.config)
-        if hasattr(parsed_args, "config")
-        else None
+        if hasattr(parsed_args, "config") and parsed_args.config
+        else {}
     )
     catalog = (
         ConcurrentDeclarativeSource.read_catalog(parsed_args.catalog)

airbyte_cdk/connector.py

@@ -31,8 +31,15 @@ def load_optional_package_file(package: str, filename: str) -> Optional[bytes]:
 
 
 class BaseConnector(ABC, Generic[TConfig]):
-    # configure whether the `check_config_against_spec_or_exit()` needs to be called
     check_config_against_spec: bool = True
+    """Configure whether `check_config_against_spec_or_exit()` needs to be called."""
+
+    check_config_during_discover: bool = True
+    """Whether `discover` requires a config and validates it against the spec.
+
+    Set to `False` for sources that can build their catalog without a config (and therefore
+    without credentials), which allows `discover` to run without `--config`.
+    """
 
     @abstractmethod
     def configure(self, config: Mapping[str, Any], temp_dir: str) -> TConfig:

airbyte_cdk/entrypoint.py

@@ -105,9 +105,11 @@ def parse_args(args: List[str]) -> argparse.Namespace:
             help="outputs a catalog describing the source's schema",
             parents=[parent_parser],
         )
-        required_discover_parser = discover_parser.add_argument_group("required named arguments")
-        required_discover_parser.add_argument(
-            "--config", type=str, required=True, help="path to the json configuration file"
+        discover_parser.add_argument(
+            "--config",
+            type=str,
+            required=False,
+            help="path to the json configuration file (optional if the source supports config-free discover)",
         )
         discover_parser.add_argument(
             "--manifest-path",
@@ -179,19 +181,38 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                 )
                 if cmd == "spec":
                     message = AirbyteMessage(type=Type.SPEC, spec=source_spec)
-                    yield from [
+                    yield from (
                         self.airbyte_message_to_string(queued_message)
                         for queued_message in self._emit_queued_messages(self.source)
-                    ]
+                    )
                     yield self.airbyte_message_to_string(message)
+                elif (
+                    cmd == "discover"
+                    and not parsed_args.config
+                    and not self.source.check_config_during_discover
+                ):
+                    yield from (
+                        self.airbyte_message_to_string(queued_message)
+                        for queued_message in self._emit_queued_messages(self.source)
+                    )
+                    yield from map(
+                        AirbyteEntrypoint.airbyte_message_to_string,
+                        self._discover_without_config(),
+                    )
+                elif cmd == "discover" and not parsed_args.config:
+                    raise AirbyteTracedException(
+                        message="This connector requires a config to run `discover`. Provide one with `--config`.",
+                        internal_message="`--config` was not provided and the source does not support config-free discover.",
+                        failure_type=FailureType.config_error,
+                    )
                 else:
                     raw_config = self.source.read_config(parsed_args.config)
                     config = self.source.configure(raw_config, temp_dir)
 
-                    yield from [
+                    yield from (
                         self.airbyte_message_to_string(queued_message)
                         for queued_message in self._emit_queued_messages(self.source)
-                    ]
+                    )
                     if cmd == "check":
                         yield from map(
                             AirbyteEntrypoint.airbyte_message_to_string,
@@ -267,6 +288,23 @@ def discover(
         yield from self._emit_queued_messages(self.source)
         yield AirbyteMessage(type=Type.CATALOG, catalog=catalog)
 
+    def _discover_without_config(self) -> Iterable[AirbyteMessage]:
+        """Run `discover` with an empty config, for sources that support config-free discover."""
+        try:
+            catalog = self.source.discover(self.logger, {})
+        except AirbyteTracedException:
+            raise
+        except Exception as error:
+            raise AirbyteTracedException(
+                message="This connector could not run `discover` without a config. Provide one with `--config`.",
+                internal_message=f"Config-free discover failed: {error!r}",
+                failure_type=FailureType.config_error,
+                exception=error,
+            ) from error
+
+        yield from self._emit_queued_messages(self.source)
+        yield AirbyteMessage(type=Type.CATALOG, catalog=catalog)
+
     def read(
         self, source_spec: ConnectorSpecification, config: TConfig, catalog: Any, state: list[Any]
     ) -> Iterable[AirbyteMessage]:

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -271,6 +271,8 @@ def __init__(
             message_repository=self._message_repository,
         )
 
+        self.check_config_during_discover = self._requires_config_for_discover()
+
     def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
         """
         Preprocesses the provided manifest dictionary by resolving any manifest references.
@@ -978,3 +980,35 @@ def _select_streams(
                     as_airbyte_message(configured_stream.stream, AirbyteStreamStatus.INCOMPLETE)
                 )
         return abstract_streams
+
+    def _requires_config_for_discover(self) -> bool:
+        """Return whether `discover` needs a config to build this source's catalog.
+
+        A config is required when the set of streams depends on the config (`ConditionalStreams`,
+        or `dynamic_streams` resolved from the API or the config rather than static parameters), or
+        when any stream schema is fetched from the API (`DynamicSchemaLoader`).
+        """
+        for dynamic_definition in self._source_config.get("dynamic_streams", []):
+            resolver_type = dynamic_definition.get("components_resolver", {}).get("type")
+            if resolver_type != "ParametrizedComponentsResolver":
+                return True
+
+        stream_definitions = [
+            self._source_config.get("streams", []),
+            self._source_config.get("dynamic_streams", []),
+        ]
+        return any(
+            _contains_component_type(stream_definitions, component_type)
+            for component_type in ("ConditionalStreams", "DynamicSchemaLoader")
+        )
+
+
+def _contains_component_type(node: Any, component_type: str) -> bool:
+    """Return whether any mapping nested under `node` has the given component `type`."""
+    if isinstance(node, Mapping):
+        return node.get("type") == component_type or any(
+            _contains_component_type(value, component_type) for value in node.values()
+        )
+    if isinstance(node, list):
+        return any(_contains_component_type(item, component_type) for item in node)
+    return False