[low-code connectors] Add request options and state to stream slicers

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/class_types_registry.py

@@ -27,6 +27,7 @@
 from airbyte_cdk.sources.declarative.stream_slicers.cartesian_product_stream_slicer import CartesianProductStreamSlicer
 from airbyte_cdk.sources.declarative.stream_slicers.datetime_stream_slicer import DatetimeStreamSlicer
 from airbyte_cdk.sources.declarative.stream_slicers.list_stream_slicer import ListStreamSlicer
+from airbyte_cdk.sources.declarative.stream_slicers.substream_slicer import SubstreamSlicer
 from airbyte_cdk.sources.declarative.transformations import RemoveFields
 from airbyte_cdk.sources.declarative.transformations.add_fields import AddFields
 
@@ -59,4 +60,5 @@
     "RecordSelector": RecordSelector,
     "RemoveFields": RemoveFields,
     "SimpleRetriever": SimpleRetriever,
+    "SubstreamSlicer": SubstreamSlicer,
 }

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/default_implementation_registry.py

@@ -30,10 +30,9 @@
 from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
 from airbyte_cdk.sources.declarative.schema.json_schema import JsonSchema
 from airbyte_cdk.sources.declarative.schema.schema_loader import SchemaLoader
-from airbyte_cdk.sources.declarative.states.dict_state import DictState
-from airbyte_cdk.sources.declarative.states.state import State
 from airbyte_cdk.sources.declarative.stream_slicers.single_slice import SingleSlice
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
+from airbyte_cdk.sources.declarative.stream_slicers.substream_slicer import ParentStreamConfig
 from airbyte_cdk.sources.streams.core import Stream
 
 """
@@ -55,8 +54,8 @@
     RequestOptionsProvider: InterpolatedRequestOptionsProvider,
     Requester: HttpRequester,
     Retriever: SimpleRetriever,
+    ParentStreamConfig: ParentStreamConfig,
     SchemaLoader: JsonSchema,
-    State: DictState,
     Stream: DeclarativeStream,
     StreamSlicer: SingleSlice,
 }

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/paginators/limit_paginator.py

@@ -127,6 +127,10 @@ def request_body_data(self) -> Mapping[str, Any]:
     def request_body_json(self) -> Mapping[str, Any]:
         return self._get_request_options(RequestOptionType.body_json)
 
+    def request_kwargs(self) -> Mapping[str, Any]:
+        # Never update kwargs
+        return {}
+
     def _get_request_options(self, option_type: RequestOptionType) -> Mapping[str, Any]:
         options = {}
         if self._page_token_option.inject_into == option_type:

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py

@@ -28,5 +28,9 @@ def request_body_data(self) -> Union[Mapping[str, Any], str]:
     def request_body_json(self) -> Mapping[str, Any]:
         return {}
 
+    def request_kwargs(self) -> Mapping[str, Any]:
+        # Never update kwargs
+        return {}
+
     def next_page_token(self, response: requests.Response, last_records: List[Mapping[str, Any]]) -> Mapping[str, Any]:
         return {}

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/paginators/paginator.py

@@ -2,13 +2,14 @@
 # Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 
-from abc import ABC, abstractmethod
+from abc import abstractmethod
 from typing import Any, List, Mapping, Optional
 
 import requests
+from airbyte_cdk.sources.declarative.requesters.request_options.request_options_provider import RequestOptionsProvider
 
 
-class Paginator(ABC):
+class Paginator(RequestOptionsProvider):
     """
     Defines the token to use to fetch the next page of records from the API.
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/request_options/request_options_provider.py

@@ -20,14 +20,13 @@ class RequestOptionsProvider(ABC):
     """
 
     @abstractmethod
-    def request_params(
-        self, stream_state: StreamState, stream_slice: Optional[StreamSlice] = None, next_page_token: Optional[Mapping[str, Any]] = None
-    ) -> MutableMapping[str, Any]:
+    def request_params(self, **kwargs) -> MutableMapping[str, Any]:
         """
         Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.
 
         E.g: you might want to define query parameters for paging if next_page_token is not None.
         """
+        pass
 
     @abstractmethod
     def request_headers(

airbyte-cdk/python/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -13,8 +13,6 @@
 from airbyte_cdk.sources.declarative.requesters.paginators.paginator import Paginator
 from airbyte_cdk.sources.declarative.requesters.requester import Requester
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
-from airbyte_cdk.sources.declarative.states.dict_state import DictState
-from airbyte_cdk.sources.declarative.states.state import State
 from airbyte_cdk.sources.declarative.stream_slicers.single_slice import SingleSlice
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 from airbyte_cdk.sources.declarative.types import Record, StreamSlice, StreamState
@@ -42,7 +40,6 @@ def __init__(
         record_selector: HttpSelector,
         paginator: Optional[Paginator] = None,
         stream_slicer: Optional[StreamSlicer] = SingleSlice(),
-        state: Optional[State] = None,
     ):
         """
         :param name: The stream's name
@@ -59,8 +56,7 @@ def __init__(
         self._requester = requester
         self._record_selector = record_selector
         super().__init__(self._requester.get_authenticator())
-        self._iterator = stream_slicer
-        self._state: State = (state or DictState()).deep_copy()
+        self._stream_slicer = stream_slicer
         self._last_response = None
         self._last_records = None
 
@@ -300,12 +296,14 @@ def read_records(
         stream_state: Optional[StreamState] = None,
     ) -> Iterable[Mapping[str, Any]]:
         # Warning: use self.state instead of the stream_state passed as argument!
+        stream_slice = stream_slice or {}  # None-check
         records_generator = HttpStream.read_records(self, sync_mode, cursor_field, stream_slice, self.state)
         for r in records_generator:
-            self._state.update_state(stream_slice=stream_slice, stream_state=self.state, last_response=self._last_response, last_record=r)
+            self._stream_slicer.update_cursor(stream_slice, last_record=r)
             yield r
         else:
-            self._state.update_state(stream_slice=stream_slice, stream_state=self.state, last_reponse=self._last_response)
+            last_record = self._last_records[-1] if self._last_records else None
+            self._stream_slicer.update_cursor(stream_slice, last_record=last_record)
             yield from []
 
     def stream_slices(
@@ -320,13 +318,13 @@ def stream_slices(
         :return:
         """
         # Warning: use self.state instead of the stream_state passed as argument!
-        return self._iterator.stream_slices(sync_mode, self.state)
+        return self._stream_slicer.stream_slices(sync_mode, self.state)
 
     @property
-    def state(self) -> StreamState:
-        return self._state.get_stream_state()
+    def state(self) -> MutableMapping[str, Any]:
+        return self._stream_slicer.get_stream_state()
 
     @state.setter
     def state(self, value: StreamState):
         """State setter, accept state serialized by state getter."""
-        self._state.set_state(value)
+        self._stream_slicer.update_cursor(value)

airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/cartesian_product_stream_slicer.py

@@ -4,7 +4,7 @@
 
 import itertools
 from collections import ChainMap
-from typing import Any, Iterable, List, Mapping
+from typing import Any, Iterable, List, Mapping, Optional
 
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
@@ -28,8 +28,34 @@ class CartesianProductStreamSlicer(StreamSlicer):
     """
 
     def __init__(self, stream_slicers: List[StreamSlicer]):
+        """
+        :param stream_slicers: Underlying stream slicers. The RequestOptions (e.g: Request headers, parameters, etc..) returned by this slicer are the combination of the RequestOptions of its input slicers. If there are conflicts e.g: two slicers define the same header or request param, the conflict is resolved by taking the value from the first slicer, where ordering is determined by the order in which slicers were input to this composite slicer.
+        """
         self._stream_slicers = stream_slicers
 
+    def update_cursor(self, stream_slice: Mapping[str, Any], last_record: Optional[Mapping[str, Any]] = None):
+        for slicer in self._stream_slicers:
+            slicer.update_cursor(stream_slice, last_record)
+
+    def request_params(self) -> Mapping[str, Any]:
+        return dict(ChainMap(*[s.request_params() for s in self._stream_slicers]))
+
+    def request_headers(self) -> Mapping[str, Any]:
+        return dict(ChainMap(*[s.request_headers() for s in self._stream_slicers]))
+
+    def request_body_data(self) -> Mapping[str, Any]:
+        return dict(ChainMap(*[s.request_body_data() for s in self._stream_slicers]))
+
+    def request_body_json(self) -> Optional[Mapping]:
+        return dict(ChainMap(*[s.request_body_json() for s in self._stream_slicers]))
+
+    def request_kwargs(self) -> Mapping[str, Any]:
+        # Never update kwargs
+        return {}
+
+    def get_stream_state(self) -> Mapping[str, Any]:
+        return dict(ChainMap(*[slicer.get_stream_state() for slicer in self._stream_slicers]))
+
     def stream_slices(self, sync_mode: SyncMode, stream_state: Mapping[str, Any]) -> Iterable[Mapping[str, Any]]:
         sub_slices = (s.stream_slices(sync_mode, stream_state) for s in self._stream_slicers)
         return (ChainMap(*a) for a in itertools.product(*sub_slices))