Merge main into devin/1769515185-thread-safe-oauth-refresh and resolve conflicts

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

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/auth/token_provider.py

@@ -71,6 +71,8 @@ def _refresh(self) -> None:
 
 @dataclass
 class InterpolatedStringTokenProvider(TokenProvider):
+    """Provides a token by interpolating a string with config values."""
+
     config: Config
     api_token: Union[InterpolatedString, str]
     parameters: Mapping[str, Any]
@@ -80,3 +82,24 @@ def __post_init__(self) -> None:
 
     def get_token(self) -> str:
         return str(self._token.eval(self.config))
+
+
+@dataclass
+class InterpolatedSessionTokenProvider(TokenProvider):
+    """Provides a token by interpolating a template with the session token.
+
+    This allows flexible token formatting, such as "Token {{ session_token }}"
+    for Django REST Framework APIs that expect "Authorization: Token <value>".
+    """
+
+    config: Config
+    api_token: Union[InterpolatedString, str]
+    session_token_provider: TokenProvider
+    parameters: Mapping[str, Any]
+
+    def __post_init__(self) -> None:
+        self._token_template = InterpolatedString.create(self.api_token, parameters=self.parameters)
+
+    def get_token(self) -> str:
+        session_token = self.session_token_provider.get_token()
+        return str(self._token_template.eval(self.config, session_token=session_token))

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -512,9 +512,16 @@ definitions:
       page_size:
         title: Page Size
         description: The number of records to include in each pages.
-        type: integer
+        anyOf:
+          - type: integer
+            title: Number of Records
+          - type: string
+            title: Interpolated Value
+        interpolation_context:
+          - config
         examples:
           - 100
+          - "{{ config['page_size'] }}"
       stop_condition:
         title: Stop Condition
         description: Template string evaluating when to stop paginating.
@@ -2067,6 +2074,18 @@ definitions:
             field_name: Authorization
           - inject_into: request_parameter
             field_name: authKey
+      api_token:
+        title: API Token Template
+        description: 'A template for the token value to inject. Use {{ session_token }} to reference the session token. For example, use "Token {{ session_token }}" for APIs that expect "Authorization: Token <token>".'
+        type: string
+        default: "{{ session_token }}"
+        interpolation_context:
+          - config
+          - session_token
+        examples:
+          - "{{ session_token }}"
+          - "Token {{ session_token }}"
+          - "Bearer {{ session_token }}"
   SessionTokenRequestBearerAuthenticator:
     title: Bearer Authenticator
     description: Authenticator for requests using the session token as a standard bearer token.
@@ -2312,13 +2331,15 @@ definitions:
           - IGNORE
           - RESET_PAGINATION
           - RATE_LIMITED
+          - REFRESH_TOKEN_THEN_RETRY
         examples:
           - SUCCESS
           - FAIL
           - RETRY
           - IGNORE
           - RESET_PAGINATION
           - RATE_LIMITED
+          - REFRESH_TOKEN_THEN_RETRY
       failure_type:
         title: Failure Type
         description: Failure type of traced exception if a response matches the filter.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -114,10 +114,10 @@ class CursorPagination(BaseModel):
         ],
         title="Cursor Value",
     )
-    page_size: Optional[int] = Field(
+    page_size: Optional[Union[int, str]] = Field(
         None,
         description="The number of records to include in each pages.",
-        examples=[100],
+        examples=[100, "{{ config['page_size'] }}"],
         title="Page Size",
     )
     stop_condition: Optional[str] = Field(
@@ -543,6 +543,7 @@ class Action(Enum):
     IGNORE = "IGNORE"
     RESET_PAGINATION = "RESET_PAGINATION"
     RATE_LIMITED = "RATE_LIMITED"
+    REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
 
 
 class FailureType(Enum):
@@ -563,6 +564,7 @@ class HttpResponseFilter(BaseModel):
             "IGNORE",
             "RESET_PAGINATION",
             "RATE_LIMITED",
+            "REFRESH_TOKEN_THEN_RETRY",
         ],
         title="Action",
     )
@@ -2057,6 +2059,16 @@ class SessionTokenRequestApiKeyAuthenticator(BaseModel):
         ],
         title="Inject API Key Into Outgoing HTTP Request",
     )
+    api_token: Optional[str] = Field(
+        "{{ session_token }}",
+        description='A template for the token value to inject. Use {{ session_token }} to reference the session token. For example, use "Token {{ session_token }}" for APIs that expect "Authorization: Token <token>".',
+        examples=[
+            "{{ session_token }}",
+            "Token {{ session_token }}",
+            "Bearer {{ session_token }}",
+        ],
+        title="API Token Template",
+    )
 
 
 class JsonSchemaPropertySelector(BaseModel):
@@ -2741,7 +2753,7 @@ class HttpRequester(BaseModelWithDeprecations):
     )
     use_cache: Optional[bool] = Field(
         False,
-        description="Enables stream requests caching. This field is automatically set by the CDK.",
+        description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).",
         title="Use Cache",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -67,6 +67,7 @@
     LegacySessionTokenAuthenticator,
 )
 from airbyte_cdk.sources.declarative.auth.token_provider import (
+    InterpolatedSessionTokenProvider,
     InterpolatedStringTokenProvider,
     SessionTokenProvider,
     TokenProvider,
@@ -1169,14 +1170,24 @@ def create_session_token_authenticator(
                 token_provider=token_provider,
             )
         else:
+            # Get the api_token template if specified, default to just the session token
+            api_token_template = (
+                getattr(model.request_authentication, "api_token", None) or "{{ session_token }}"
+            )
+            final_token_provider: TokenProvider = InterpolatedSessionTokenProvider(
+                config=config,
+                api_token=api_token_template,
+                session_token_provider=token_provider,
+                parameters=model.parameters or {},
+            )
             return self.create_api_key_authenticator(
                 ApiKeyAuthenticatorModel(
                     type="ApiKeyAuthenticator",
                     api_token="",
                     inject_into=model.request_authentication.inject_into,
                 ),  # type: ignore # $parameters and headers default to None
                 config=config,
-                token_provider=token_provider,
+                token_provider=final_token_provider,
             )
 
     @staticmethod
@@ -1554,14 +1565,18 @@ def create_concurrent_cursor_from_incrementing_count_cursor(
                 f"Expected {model_type.__name__} component, but received {incrementing_count_cursor_model.__class__.__name__}"
             )
 
-        interpolated_start_value = (
-            InterpolatedString.create(
-                incrementing_count_cursor_model.start_value,  # type: ignore
+        start_value: Union[int, str, None] = incrementing_count_cursor_model.start_value
+        # Pydantic Union type coercion can convert int 0 to string '0' depending on Union order.
+        # We need to handle both int and str representations of numeric values.
+        # Evaluate the InterpolatedString and convert to int for the ConcurrentCursor.
+        if start_value is not None:
+            interpolated_start_value = InterpolatedString.create(
+                str(start_value),  # Ensure we pass a string to InterpolatedString.create
                 parameters=incrementing_count_cursor_model.parameters or {},
             )
-            if incrementing_count_cursor_model.start_value
-            else 0
-        )
+            evaluated_start_value: int = int(interpolated_start_value.eval(config=config))
+        else:
+            evaluated_start_value = 0
 
         cursor_field = self._get_catalog_defined_cursor_field(
             stream_name=stream_name,
@@ -1593,7 +1608,7 @@ def create_concurrent_cursor_from_incrementing_count_cursor(
             connector_state_converter=connector_state_converter,
             cursor_field=cursor_field,
             slice_boundary_fields=None,
-            start=interpolated_start_value,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
+            start=evaluated_start_value,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
             end_provider=connector_state_converter.get_end_provider(),  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
         )
 
@@ -1745,10 +1760,16 @@ def create_cursor_pagination(
                 self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(inner_decoder))
             )
 
+        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
+        # If page_size is a string that represents an integer (not an interpolation), convert it back.
+        page_size = model.page_size
+        if isinstance(page_size, str) and page_size.isdigit():
+            page_size = int(page_size)
+
         return CursorPaginationStrategy(
             cursor_value=model.cursor_value,
             decoder=decoder_to_use,
-            page_size=model.page_size,
+            page_size=page_size,
             stop_condition=model.stop_condition,
             config=config,
             parameters=model.parameters or {},
@@ -2917,8 +2938,14 @@ def create_offset_increment(
             else None
         )
 
+        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
+        # If page_size is a string that represents an integer (not an interpolation), convert it back.
+        page_size = model.page_size
+        if isinstance(page_size, str) and page_size.isdigit():
+            page_size = int(page_size)
+
         return OffsetIncrement(
-            page_size=model.page_size,
+            page_size=page_size,
             config=config,
             decoder=decoder_to_use,
             extractor=extractor,
@@ -2930,8 +2957,14 @@ def create_offset_increment(
     def create_page_increment(
         model: PageIncrementModel, config: Config, **kwargs: Any
     ) -> PageIncrement:
+        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
+        # If page_size is a string that represents an integer (not an interpolation), convert it back.
+        page_size = model.page_size
+        if isinstance(page_size, str) and page_size.isdigit():
+            page_size = int(page_size)
+
         return PageIncrement(
-            page_size=model.page_size,
+            page_size=page_size,
             config=config,
             start_from_page=model.start_from_page or 0,
             inject_on_first_request=model.inject_on_first_request or False,

airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py

@@ -23,10 +23,10 @@
 @dataclass
 class CursorPaginationStrategy(PaginationStrategy):
     """
-    Pagination strategy that evaluates an interpolated string to define the next page token
+    Pagination strategy that evaluates an interpolated string to define the next page token.
 
     Attributes:
-        page_size (Optional[int]): the number of records to request
+        page_size (Optional[Union[str, int]]): the number of records to request
         cursor_value (Union[InterpolatedString, str]): template string evaluating to the cursor value
         config (Config): connection config
         stop_condition (Optional[InterpolatedBoolean]): template string evaluating when to stop paginating
@@ -36,7 +36,7 @@ class CursorPaginationStrategy(PaginationStrategy):
     cursor_value: Union[InterpolatedString, str]
     config: Config
     parameters: InitVar[Mapping[str, Any]]
-    page_size: Optional[int] = None
+    page_size: Optional[Union[str, int]] = None
     stop_condition: Optional[Union[InterpolatedBoolean, str]] = None
     decoder: Decoder = field(
         default_factory=lambda: PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
@@ -54,6 +54,14 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         else:
             self._stop_condition = self.stop_condition
 
+        if isinstance(self.page_size, int) or (self.page_size is None):
+            self._page_size = self.page_size
+        else:
+            page_size = InterpolatedString(self.page_size, parameters=parameters).eval(self.config)
+            if not isinstance(page_size, int):
+                raise Exception(f"{page_size} is of type {type(page_size)}. Expected {int}")
+            self._page_size = page_size
+
     @property
     def initial_token(self) -> Optional[Any]:
         """
@@ -95,4 +103,4 @@ def next_page_token(
         return token if token else None
 
     def get_page_size(self) -> Optional[int]:
-        return self.page_size
+        return self._page_size

airbyte_cdk/sources/file_based/config/abstract_file_based_spec.py

@@ -7,12 +7,13 @@
 from typing import Any, Dict, List, Literal, Optional, Union
 
 import dpath
-from pydantic.v1 import AnyUrl, BaseModel, Field
+from pydantic.v1 import AnyUrl, BaseModel, Field, validator
 
 from airbyte_cdk import OneOfOptionConfig
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig
 from airbyte_cdk.sources.specs.transfer_modes import DeliverPermissions
 from airbyte_cdk.sources.utils import schema_helpers
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_try_parse
 
 
 class DeliverRecords(BaseModel):
@@ -53,13 +54,39 @@ class AbstractFileBasedSpec(BaseModel):
     start_date: Optional[str] = Field(
         title="Start Date",
         description="UTC date and time in the format 2017-01-25T00:00:00.000000Z. Any file modified before this date will not be replicated.",
-        examples=["2021-01-01T00:00:00.000000Z"],
+        examples=[
+            "2021-01-01",
+            "2021-01-01T00:00:00Z",
+            "2021-01-01T00:00:00.000Z",
+            "2021-01-01T00:00:00.000000Z",
+        ],
         format="date-time",
-        pattern="^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{6}Z$",
-        pattern_descriptor="YYYY-MM-DDTHH:mm:ss.SSSSSSZ",
+        pattern=r"^[0-9]{4}-[0-9]{2}-[0-9]{2}(T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:[0-9]{2})?)?$",
+        pattern_descriptor="YYYY-MM-DD, YYYY-MM-DDTHH:mm:ssZ, or YYYY-MM-DDTHH:mm:ss.SSSSSSZ",
         order=1,
     )
 
+    @validator("start_date", pre=True)
+    def validate_start_date(
+        cls,  # noqa: N805  # Pydantic validators use cls, not self
+        v: Optional[str],
+    ) -> Optional[str]:
+        """Validate that start_date is a parseable datetime string.
+
+        Uses ab_datetime_try_parse which accepts any common ISO8601/RFC3339 format,
+        including formats with or without microseconds (e.g., both
+        '2021-01-01T00:00:00Z' and '2021-01-01T00:00:00.000000Z' are valid).
+        """
+        if v is None:
+            return v
+        parsed = ab_datetime_try_parse(v)
+        if parsed is None:
+            raise ValueError(
+                f"'{v}' is not a valid datetime string. "
+                "Please use a format like '2021-01-01T00:00:00Z' or '2021-01-01T00:00:00.000000Z'."
+            )
+        return v
+
     streams: List[FileBasedStreamConfig] = Field(
         title="The list of streams to sync",
         description='Each instance of this configuration defines a <a href="https://docs.airbyte.com/cloud/core-concepts#stream">stream</a>. Use this to define which files belong in the stream, their format, and how they should be parsed and validated. When sending data to warehouse destination such as Snowflake or BigQuery, each stream is a separate table.',

airbyte_cdk/sources/streams/http/error_handlers/response_models.py

@@ -18,6 +18,7 @@ class ResponseAction(Enum):
     IGNORE = "IGNORE"
     RESET_PAGINATION = "RESET_PAGINATION"
     RATE_LIMITED = "RATE_LIMITED"
+    REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
 
 
 @dataclass