Merge branch 'main' into pedro/add-manifest-runner

Author: pedroslopez

.github/pr-welcome-internal.md

@@ -26,6 +26,7 @@ Airbyte Maintainers can execute the following slash commands on your PR:
 - `/autofix` - Fixes most formatting and linting issues
 - `/poetry-lock` - Updates poetry.lock file
 - `/test` - Runs connector tests with the updated CDK
+- `/poe build` - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
 - `/poe <command>` - Runs any poe command in the CDK environment
 
 [📝 _Edit this welcome message._](https://github.com/airbytehq/airbyte-python-cdk/blob/main/.github/pr-welcome-internal.md)

airbyte_cdk/cli/airbyte_cdk/_connector.py

@@ -123,11 +123,18 @@ def connector_cli_group() -> None:
     multiple=True,
     help="Additional argument(s) to pass to pytest. Can be specified multiple times.",
 )
+@click.option(
+    "--no-creds",
+    is_flag=True,
+    default=False,
+    help="Skip tests that require credentials (marked with 'requires_creds').",
+)
 def connector_test(
     connector: str | Path | None = None,
     *,
     collect_only: bool = False,
     pytest_args: list[str] | None = None,
+    no_creds: bool = False,
 ) -> None:
     """Run connector tests.
 
@@ -147,6 +154,9 @@ def connector_test(
     if collect_only:
         pytest_args.append("--collect-only")
 
+    if no_creds:
+        pytest_args.extend(["-m", "not requires_creds"])
+
     run_connector_tests(
         connector_name=connector_name,
         connector_directory=connector_directory,

airbyte_cdk/cli/airbyte_cdk/_image.py

@@ -100,10 +100,17 @@ def build(
     "--image",
     help="Image to test, instead of building a new one.",
 )
+@click.option(
+    "--no-creds",
+    is_flag=True,
+    default=False,
+    help="Skip tests that require credentials (marked with 'requires_creds').",
+)
 def image_test(  # "image test" command
     connector: str | None = None,
     *,
     image: str | None = None,
+    no_creds: bool = False,
 ) -> None:
     """Test a connector Docker image.
 
@@ -124,7 +131,11 @@ def image_test(  # "image test" command
     connector_name, connector_directory = resolve_connector_name_and_directory(connector)
 
     # Select only tests with the 'image_tests' mark
-    pytest_args = ["-m", "image_tests"]
+    pytest_filter = "image_tests"
+    if no_creds:
+        pytest_filter += " and not requires_creds"
+
+    pytest_args = ["-m", pytest_filter]
     if not image:
         metadata_file_path: Path = connector_directory / "metadata.yaml"
         try:

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -108,7 +108,12 @@ def read_stream(
         stream_name = configured_catalog.streams[0].stream.name
 
         stream_read = test_read_handler.run_test_read(
-            source, config, configured_catalog, state, limits.max_records
+            source,
+            config,
+            configured_catalog,
+            stream_name,
+            state,
+            limits.max_records,
         )
 
         return AirbyteMessage(

airbyte_cdk/connector_builder/test_reader/helpers.py

@@ -269,6 +269,37 @@ def should_close_page_for_slice(at_least_one_page_in_group: bool, message: Airby
     return at_least_one_page_in_group and should_process_slice_descriptor(message)
 
 
+def is_page_http_request_for_different_stream(
+    json_message: Optional[Dict[str, Any]], stream_name: str
+) -> bool:
+    """
+    Determines whether a given JSON message represents a page HTTP request for a different stream.
+
+    This function checks if the provided JSON message is a page HTTP request, and if the stream name in the log is
+    different from the provided stream name.
+
+    This is needed because dynamic streams result in extra page HTTP requests for the dynamic streams that we want to ignore
+    when they do not match the stream that is being read.
+
+    Args:
+        json_message (Optional[Dict[str, Any]]): The JSON message to evaluate.
+        stream_name (str): The name of the stream to compare against.
+
+    Returns:
+        bool: True if the JSON message is a page HTTP request for a different stream, False otherwise.
+    """
+    if not json_message or not is_page_http_request(json_message):
+        return False
+
+    message_stream_name: str | None = (
+        json_message.get("airbyte_cdk", {}).get("stream", {}).get("name", None)
+    )
+    if message_stream_name is None:
+        return False
+
+    return message_stream_name != stream_name
+
+
 def is_page_http_request(json_message: Optional[Dict[str, Any]]) -> bool:
     """
     Determines whether a given JSON message represents a page HTTP request.

airbyte_cdk/connector_builder/test_reader/message_grouper.py

@@ -28,6 +28,7 @@
     is_async_auxiliary_request,
     is_config_update_message,
     is_log_message,
+    is_page_http_request_for_different_stream,
     is_record_message,
     is_state_message,
     is_trace_with_error,
@@ -44,6 +45,7 @@ def get_message_groups(
     schema_inferrer: SchemaInferrer,
     datetime_format_inferrer: DatetimeFormatInferrer,
     limit: int,
+    stream_name: str,
 ) -> MESSAGE_GROUPS:
     """
     Processes an iterator of AirbyteMessage objects to group and yield messages based on their type and sequence.
@@ -96,6 +98,9 @@ def get_message_groups(
     while records_count < limit and (message := next(messages, None)):
         json_message = airbyte_message_to_json(message)
 
+        if is_page_http_request_for_different_stream(json_message, stream_name):
+            continue
+
         if should_close_page(at_least_one_page_in_group, message, json_message):
             current_page_request, current_page_response = handle_current_page(
                 current_page_request,

airbyte_cdk/connector_builder/test_reader/reader.py

@@ -86,6 +86,7 @@ def run_test_read(
         source: DeclarativeSource,
         config: Mapping[str, Any],
         configured_catalog: ConfiguredAirbyteCatalog,
+        stream_name: str,
         state: List[AirbyteStateMessage],
         record_limit: Optional[int] = None,
     ) -> StreamRead:
@@ -112,14 +113,21 @@ def run_test_read(
 
         record_limit = self._check_record_limit(record_limit)
         # The connector builder currently only supports reading from a single stream at a time
-        stream = source.streams(config)[0]
+        streams = source.streams(config)
+        stream = next((stream for stream in streams if stream.name == stream_name), None)
 
         # get any deprecation warnings during the component creation
         deprecation_warnings: List[LogMessage] = source.deprecation_warnings()
 
         schema_inferrer = SchemaInferrer(
-            self._pk_to_nested_and_composite_field(stream.primary_key),
-            self._cursor_field_to_nested_and_composite_field(stream.cursor_field),
+            self._pk_to_nested_and_composite_field(
+                stream.primary_key if hasattr(stream, "primary_key") else stream._primary_key  # type: ignore  # We are accessing the private property here as the primary key is not exposed. We should either expose it or use `as_airbyte_stream` to retrieve it as this is the "official" way where it is exposed in the Airbyte protocol
+            )
+            if stream
+            else None,
+            self._cursor_field_to_nested_and_composite_field(stream.cursor_field)
+            if stream
+            else None,
         )
         datetime_format_inferrer = DatetimeFormatInferrer()
 
@@ -128,6 +136,7 @@ def run_test_read(
             schema_inferrer,
             datetime_format_inferrer,
             record_limit,
+            stream_name,
         )
 
         slices, log_messages, auxiliary_requests, latest_config_update = self._categorise_groups(

airbyte_cdk/entrypoint.py

@@ -22,7 +22,7 @@
 
 from airbyte_cdk.connector import TConfig
 from airbyte_cdk.exception_handler import init_uncaught_exception_handler
-from airbyte_cdk.logger import PRINT_BUFFER, init_logger
+from airbyte_cdk.logger import PRINT_BUFFER, init_logger, is_platform_debug_log_enabled
 from airbyte_cdk.models import (
     AirbyteConnectionStatus,
     AirbyteMessage,
@@ -158,7 +158,7 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
         if not cmd:
             raise Exception("No command passed")
 
-        if hasattr(parsed_args, "debug") and parsed_args.debug:
+        if (hasattr(parsed_args, "debug") and parsed_args.debug) or is_platform_debug_log_enabled():
             self.logger.setLevel(logging.DEBUG)
             logger.setLevel(logging.DEBUG)
             self.logger.debug("Debug logs enabled")

airbyte_cdk/logger.py

@@ -1,10 +1,10 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-
 import json
 import logging
 import logging.config
+import os
 from typing import Any, Callable, Mapping, Optional, Tuple
 
 import orjson
@@ -40,6 +40,10 @@
 }
 
 
+def is_platform_debug_log_enabled() -> bool:
+    return os.environ.get("LOG_LEVEL", "info").lower() == "debug"
+
+
 def init_logger(name: Optional[str] = None) -> logging.Logger:
     """Initial set up of logger"""
     logger = logging.getLogger(name)
@@ -73,8 +77,22 @@ def format(self, record: logging.LogRecord) -> str:
         airbyte_level = self.level_mapping.get(record.levelno, "INFO")
         if airbyte_level == Level.DEBUG:
             extras = self.extract_extra_args_from_record(record)
-            debug_dict = {"type": "DEBUG", "message": record.getMessage(), "data": extras}
-            return filter_secrets(json.dumps(debug_dict))
+            if is_platform_debug_log_enabled():
+                # We have a different behavior between debug logs enabled through `--debug` argument and debug logs
+                # enabled through environment variable. The reason is that for platform logs, we need to have these
+                # printed as AirbyteMessage which is not the case with the current previous implementation.
+                # Why not migrate both to AirbyteMessages then? AirbyteMessages do not support having structured logs.
+                # which means that the DX would be degraded compared to the current solution (devs will need to identify
+                # the `log.message` field and figure out where in this field is the response while the current solution
+                # have a specific field that is structured for extras.
+                message = f"{filter_secrets(record.getMessage())} ///\nExtra logs: {filter_secrets(json.dumps(extras))}"
+                log_message = AirbyteMessage(
+                    type=Type.LOG, log=AirbyteLogMessage(level=airbyte_level, message=message)
+                )
+                return orjson.dumps(AirbyteMessageSerializer.dump(log_message)).decode()
+            else:
+                debug_dict = {"type": "DEBUG", "message": record.getMessage(), "data": extras}
+                return filter_secrets(json.dumps(debug_dict))
         else:
             message = super().format(record)
             message = filter_secrets(message)

airbyte_cdk/sources/declarative/checks/check_dynamic_stream.py

@@ -3,12 +3,13 @@
 #
 
 import logging
-import traceback
 from dataclasses import InitVar, dataclass
-from typing import Any, List, Mapping, Tuple
+from typing import Any, List, Mapping, Tuple, Union
 
-from airbyte_cdk import AbstractSource
+from airbyte_cdk.sources.abstract_source import AbstractSource
+from airbyte_cdk.sources.declarative.checks.check_stream import evaluate_availability
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
+from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 
 
@@ -34,20 +35,16 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
     def check_connection(
         self, source: AbstractSource, logger: logging.Logger, config: Mapping[str, Any]
     ) -> Tuple[bool, Any]:
-        streams = source.streams(config=config)
+        streams: List[Union[Stream, AbstractStream]] = source.streams(config=config)  # type: ignore  # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
 
         if len(streams) == 0:
             return False, f"No streams to connect to from source {source}"
         if not self.use_check_availability:
             return True, None
 
-        availability_strategy = HttpAvailabilityStrategy()
-
         try:
             for stream in streams[: min(self.stream_count, len(streams))]:
-                stream_is_available, reason = availability_strategy.check_availability(
-                    stream, logger
-                )
+                stream_is_available, reason = evaluate_availability(stream, logger)
                 if not stream_is_available:
                     logger.warning(f"Stream {stream.name} is not available: {reason}")
                     return False, reason