Merge branch 'main' into daryna/low-code/refactor-model-to-component-factory

Author: darynaishchenko

.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/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

airbyte_cdk/sources/declarative/checks/check_stream.py

@@ -5,13 +5,30 @@
 import logging
 import traceback
 from dataclasses import InitVar, dataclass
-from typing import Any, Dict, List, Mapping, Optional, Tuple
+from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
 
-from airbyte_cdk import AbstractSource
+from airbyte_cdk.sources.abstract_source import AbstractSource
 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.core import Stream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 
 
+def evaluate_availability(
+    stream: Union[Stream, AbstractStream], logger: logging.Logger
+) -> Tuple[bool, Optional[str]]:
+    """
+    As a transition period, we want to support both Stream and AbstractStream until we migrate everything to AbstractStream.
+    """
+    if isinstance(stream, Stream):
+        return HttpAvailabilityStrategy().check_availability(stream, logger)
+    elif isinstance(stream, AbstractStream):
+        availability = stream.check_availability()
+        return availability.is_available, availability.reason
+    else:
+        raise ValueError(f"Unsupported stream type {type(stream)}")
+
+
 @dataclass(frozen=True)
 class DynamicStreamCheckConfig:
     """Defines the configuration for dynamic stream during connection checking. This class specifies
@@ -51,7 +68,7 @@ def check_connection(
     ) -> Tuple[bool, Any]:
         """Checks the connection to the source and its streams."""
         try:
-            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 not streams:
                 return False, f"No streams to connect to from source {source}"
         except Exception as error:
@@ -82,13 +99,15 @@ def check_connection(
         return True, None
 
     def _check_stream_availability(
-        self, stream_name_to_stream: Dict[str, Any], stream_name: str, logger: logging.Logger
+        self,
+        stream_name_to_stream: Dict[str, Union[Stream, AbstractStream]],
+        stream_name: str,
+        logger: logging.Logger,
     ) -> Tuple[bool, Any]:
         """Checks if streams are available."""
-        availability_strategy = HttpAvailabilityStrategy()
         try:
             stream = stream_name_to_stream[stream_name]
-            stream_is_available, reason = availability_strategy.check_availability(stream, logger)
+            stream_is_available, reason = evaluate_availability(stream, logger)
             if not stream_is_available:
                 message = f"Stream {stream_name} is not available: {reason}"
                 logger.warning(message)
@@ -98,7 +117,10 @@ def _check_stream_availability(
         return True, None
 
     def _check_dynamic_streams_availability(
-        self, source: AbstractSource, stream_name_to_stream: Dict[str, Any], logger: logging.Logger
+        self,
+        source: AbstractSource,
+        stream_name_to_stream: Dict[str, Union[Stream, AbstractStream]],
+        logger: logging.Logger,
     ) -> Tuple[bool, Any]:
         """Checks the availability of dynamic streams."""
         dynamic_streams = source.resolved_manifest.get("dynamic_streams", [])  # type: ignore[attr-defined] # The source's resolved_manifest manifest is checked before calling this method
@@ -135,18 +157,15 @@ def _map_generated_streams(
     def _check_generated_streams_availability(
         self,
         generated_streams: List[Dict[str, Any]],
-        stream_name_to_stream: Dict[str, Any],
+        stream_name_to_stream: Dict[str, Union[Stream, AbstractStream]],
         logger: logging.Logger,
         max_count: int,
     ) -> Tuple[bool, Any]:
         """Checks availability of generated dynamic streams."""
-        availability_strategy = HttpAvailabilityStrategy()
         for declarative_stream in generated_streams[: min(max_count, len(generated_streams))]:
             stream = stream_name_to_stream[declarative_stream["name"]]
             try:
-                stream_is_available, reason = availability_strategy.check_availability(
-                    stream, logger
-                )
+                stream_is_available, reason = evaluate_availability(stream, logger)
                 if not stream_is_available:
                     message = f"Dynamic Stream {stream.name} is not available: {reason}"
                     logger.warning(message)

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -52,9 +52,6 @@
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
 from airbyte_cdk.sources.streams.concurrent.abstract_stream_facade import AbstractStreamFacade
-from airbyte_cdk.sources.streams.concurrent.availability_strategy import (
-    AlwaysAvailableAvailabilityStrategy,
-)
 from airbyte_cdk.sources.streams.concurrent.cursor import ConcurrentCursor, FinalStateCursor
 from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream
 from airbyte_cdk.sources.streams.concurrent.helpers import get_primary_key_from_stream
@@ -325,7 +322,6 @@ def _group_streams(
                             partition_generator=partition_generator,
                             name=declarative_stream.name,
                             json_schema=declarative_stream.get_json_schema(),
-                            availability_strategy=AlwaysAvailableAvailabilityStrategy(),
                             primary_key=get_primary_key_from_stream(declarative_stream.primary_key),
                             cursor_field=cursor.cursor_field.cursor_field_key
                             if hasattr(cursor, "cursor_field")
@@ -362,7 +358,6 @@ def _group_streams(
                             partition_generator=partition_generator,
                             name=declarative_stream.name,
                             json_schema=declarative_stream.get_json_schema(),
-                            availability_strategy=AlwaysAvailableAvailabilityStrategy(),
                             primary_key=get_primary_key_from_stream(declarative_stream.primary_key),
                             cursor_field=None,
                             logger=self.logger,
@@ -417,7 +412,6 @@ def _group_streams(
                             partition_generator=partition_generator,
                             name=declarative_stream.name,
                             json_schema=declarative_stream.get_json_schema(),
-                            availability_strategy=AlwaysAvailableAvailabilityStrategy(),
                             primary_key=get_primary_key_from_stream(declarative_stream.primary_key),
                             cursor_field=perpartition_cursor.cursor_field.cursor_field_key,
                             logger=self.logger,

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1838,6 +1838,7 @@ definitions:
         "$ref": "#/definitions/RequestOption"
       page_token_option:
         title: Inject Page Token Into Outgoing HTTP Request
+        description: Inject the page token into the outgoing HTTP requests by inserting it into either the request URL path or a field on the request.
         anyOf:
           - "$ref": "#/definitions/RequestOption"
           - "$ref": "#/definitions/RequestPath"
@@ -3486,7 +3487,7 @@ definitions:
           - [["content", "html"], ["content", "plain_text"]]
   RequestPath:
     title: Request Path
-    description: Specifies where in the request path a component's value should be inserted.
+    description: The URL path to be used for the HTTP request.
     type: object
     required:
       - type

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2002,7 +2002,9 @@ class DefaultPaginator(BaseModel):
         None, title="Inject Page Size Into Outgoing HTTP Request"
     )
     page_token_option: Optional[Union[RequestOption, RequestPath]] = Field(
-        None, title="Inject Page Token Into Outgoing HTTP Request"
+        None,
+        description="Inject the page token into the outgoing HTTP requests by inserting it into either the request URL path or a field on the request.",
+        title="Inject Page Token Into Outgoing HTTP Request",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 

airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -168,7 +168,13 @@ def _get_url(
             next_page_token=next_page_token,
         )
 
-        full_url = self._join_url(url_base, path) if url_base else url + path if path else url
+        full_url = (
+            self._join_url(url_base, path)
+            if url_base
+            else self._join_url(url, path)
+            if path
+            else url
+        )
 
         return full_url
 

airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py

@@ -159,14 +159,7 @@ def path(
     ) -> Optional[str]:
         token = next_page_token.get("next_page_token") if next_page_token else None
         if token and self.page_token_option and isinstance(self.page_token_option, RequestPath):
-            # make additional interpolation context
-            interpolation_context = get_interpolation_context(
-                stream_state=stream_state,
-                stream_slice=stream_slice,
-                next_page_token=next_page_token,
-            )
-            # Replace url base to only return the path
-            return str(token).replace(self.url_base.eval(self.config, **interpolation_context), "")  # type: ignore # url_base is casted to a InterpolatedString in __post_init__
+            return str(token)
         else:
             return None