feat: Removes stream_state interpolation from CDK

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -80,7 +80,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - "{{ record['updates'] }}"
           - "{{ record['MetaData']['LastUpdatedTime'] }}"
@@ -794,7 +793,7 @@ definitions:
         description: This option is used to adjust the upper and lower boundaries of each datetime window to beginning and end of the provided target period (day, week, month)
         type: object
         required:
-        - target
+          - target
         properties:
           target:
             title: Target
@@ -1611,7 +1610,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - "/products"
           - "/quotes/{{ stream_partition['id'] }}/quote_line_groups"
@@ -1661,7 +1659,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - |
             [{"clause": {"type": "timestamp", "operator": 10, "parameters":
@@ -1679,7 +1676,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - sort_order: "ASC"
             sort_field: "CREATED_AT"
@@ -1700,7 +1696,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - Output-Format: JSON
           - Version: "{{ config['version'] }}"
@@ -1717,7 +1712,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - unit: "day"
           - query: 'last_event_time BETWEEN TIMESTAMP "{{ stream_interval.start_time }}" AND TIMESTAMP "{{ stream_interval.end_time }}"'
@@ -2072,7 +2066,6 @@ definitions:
         interpolation_context:
           - config
           - record
-          - stream_state
           - stream_slice
       new:
         type: string
@@ -2086,7 +2079,6 @@ definitions:
         interpolation_context:
           - config
           - record
-          - stream_state
           - stream_slice
       $parameters:
         type: object
@@ -2753,7 +2745,6 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
-          - stream_state
         examples:
           - "{{ record['created_at'] >= stream_interval['start_time'] }}"
           - "{{ record.status in ['active', 'expired'] }}"

airbyte_cdk/sources/declarative/interpolation/jinja.py

@@ -15,6 +15,8 @@
 from airbyte_cdk.sources.declarative.interpolation.interpolation import Interpolation
 from airbyte_cdk.sources.declarative.interpolation.macros import macros
 from airbyte_cdk.sources.types import Config
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.utils import AirbyteTracedException
 
 
 class StreamPartitionAccessEnvironment(SandboxedEnvironment):
@@ -36,6 +38,10 @@ def is_safe_attribute(self, obj: Any, attr: str, value: Any) -> bool:
     "stream_partition": "stream_slice",  # Use stream_partition to access partition router's values
 }
 
+_UNSUPPORTED_INTERPOLATION_VARIABLES: Mapping[str, str] = {
+    "stream_state": "`stream_state` is no longer supported for interpolation. We recommend using `stream_interval` instead. Please reference the CDK Migration Guide for more information.",
+}
+
 # These extensions are not installed so they're not currently a problem,
 # but we're still explicitly removing them from the jinja context.
 # At worst, this is documentation that we do NOT want to include these extensions because of the potential security risks
@@ -95,6 +101,13 @@ def eval(
             elif equivalent in context:
                 context[alias] = context[equivalent]
 
+        for variable_name in _UNSUPPORTED_INTERPOLATION_VARIABLES:
+            if variable_name in input_str:
+                raise AirbyteTracedException(
+                    message=_UNSUPPORTED_INTERPOLATION_VARIABLES[variable_name],
+                    internal_message=_UNSUPPORTED_INTERPOLATION_VARIABLES[variable_name],
+                    failure_type=FailureType.config_error,
+                )
         try:
             if isinstance(input_str, str):
                 result = self._eval(input_str, context)

airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -120,7 +120,6 @@ def get_path(
         next_page_token: Optional[Mapping[str, Any]],
     ) -> str:
         kwargs = {
-            "stream_state": stream_state,
             "stream_slice": stream_slice,
             "next_page_token": next_page_token,
         }

airbyte_cdk/sources/declarative/requesters/request_options/interpolated_request_input_provider.py

@@ -37,7 +37,6 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 
     def eval_request_inputs(
         self,
-        stream_state: Optional[StreamState] = None,
         stream_slice: Optional[StreamSlice] = None,
         next_page_token: Optional[Mapping[str, Any]] = None,
         valid_key_types: Optional[Tuple[Type[Any]]] = None,
@@ -54,7 +53,6 @@ def eval_request_inputs(
         :return: The request inputs to set on an outgoing HTTP request
         """
         kwargs = {
-            "stream_state": stream_state,
             "stream_slice": stream_slice,
             "next_page_token": next_page_token,
         }

airbyte_cdk/sources/declarative/requesters/request_options/interpolated_request_options_provider.py

@@ -5,8 +5,6 @@
 from dataclasses import InitVar, dataclass, field
 from typing import Any, Mapping, MutableMapping, Optional, Union
 
-from typing_extensions import deprecated
-
 from airbyte_cdk.sources.declarative.interpolation.interpolated_nested_mapping import NestedMapping
 from airbyte_cdk.sources.declarative.requesters.request_options.interpolated_nested_request_input_provider import (
     InterpolatedNestedRequestInputProvider,
@@ -17,7 +15,6 @@
 from airbyte_cdk.sources.declarative.requesters.request_options.request_options_provider import (
     RequestOptionsProvider,
 )
-from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
 RequestInput = Union[str, Mapping[str, str]]
@@ -80,7 +77,6 @@ def get_request_params(
         next_page_token: Optional[Mapping[str, Any]] = None,
     ) -> MutableMapping[str, Any]:
         interpolated_value = self._parameter_interpolator.eval_request_inputs(
-            stream_state,
             stream_slice,
             next_page_token,
             valid_key_types=(str,),
@@ -98,7 +94,7 @@ def get_request_headers(
         next_page_token: Optional[Mapping[str, Any]] = None,
     ) -> Mapping[str, Any]:
         return self._headers_interpolator.eval_request_inputs(
-            stream_state, stream_slice, next_page_token
+            stream_slice, next_page_token
         )
 
     def get_request_body_data(
@@ -109,7 +105,6 @@ def get_request_body_data(
         next_page_token: Optional[Mapping[str, Any]] = None,
     ) -> Union[Mapping[str, Any], str]:
         return self._body_data_interpolator.eval_request_inputs(
-            stream_state,
             stream_slice,
             next_page_token,
             valid_key_types=(str,),
@@ -126,39 +121,3 @@ def get_request_body_json(
         return self._body_json_interpolator.eval_request_inputs(
             stream_state, stream_slice, next_page_token
         )
-
-    @deprecated(
-        "This class is temporary and used to incrementally deliver low-code to concurrent",
-        category=ExperimentalClassWarning,
-    )
-    def request_options_contain_stream_state(self) -> bool:
-        """
-        Temporary helper method used as we move low-code streams to the concurrent framework. This method determines if
-        the InterpolatedRequestOptionsProvider has is a dependency on a non-thread safe interpolation context such as
-        stream_state.
-        """
-
-        return (
-            self._check_if_interpolation_uses_stream_state(self.request_parameters)
-            or self._check_if_interpolation_uses_stream_state(self.request_headers)
-            or self._check_if_interpolation_uses_stream_state(self.request_body_data)
-            or self._check_if_interpolation_uses_stream_state(self.request_body_json)
-        )
-
-    @staticmethod
-    def _check_if_interpolation_uses_stream_state(
-        request_input: Optional[Union[RequestInput, NestedMapping]],
-    ) -> bool:
-        if not request_input:
-            return False
-        elif isinstance(request_input, str):
-            return "stream_state" in request_input
-        else:
-            for key, val in request_input.items():
-                # Covers the case of RequestInput in the form of a string or Mapping[str, str]. It also covers the case
-                # of a NestedMapping where the value is a string.
-                # Note: Doesn't account for nested mappings for request_body_json, but I don't see stream_state used in that way
-                # in our code
-                if "stream_state" in key or (isinstance(val, str) and "stream_state" in val):
-                    return True
-        return False

airbyte_cdk/sources/declarative/transformations/add_fields.py

@@ -128,7 +128,7 @@ def transform(
     ) -> None:
         if config is None:
             config = {}
-        kwargs = {"record": record, "stream_state": stream_state, "stream_slice": stream_slice}
+        kwargs = {"record": record, "stream_slice": stream_slice}
         for parsed_field in self._parsed_fields:
             valid_types = (parsed_field.value_type,) if parsed_field.value_type else None
             value = parsed_field.value.eval(config, valid_types=valid_types, **kwargs)

unit_tests/sources/declarative/requesters/test_interpolated_request_input_provider.py

@@ -45,7 +45,7 @@ def test_initialize_interpolated_mapping_request_input_provider(
     provider = InterpolatedRequestInputProvider(
         request_inputs=input_request_data, config=config, parameters=parameters
     )
-    actual_request_data = provider.eval_request_inputs(stream_state={}, stream_slice=stream_slice)
+    actual_request_data = provider.eval_request_inputs(stream_slice=stream_slice)
 
     assert isinstance(provider._interpolator, InterpolatedMapping)
     assert actual_request_data == expected_request_data