Remove RFR stuff

Author: maxime.c

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -84,7 +84,6 @@ def __init__(
         # incremental streams running in full refresh.
         component_factory = component_factory or ModelToComponentFactory(
             emit_connector_builder_messages=emit_connector_builder_messages,
-            disable_resumable_full_refresh=True,
             connector_state_manager=self._connector_state_manager,
             max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
         )

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -7,6 +7,7 @@
 import datetime
 import importlib
 import inspect
+import logging
 import re
 from functools import partial
 from typing import (
@@ -544,6 +545,8 @@
     StreamSlicer,
     StreamSlicerTestReadDecorator,
 )
+from airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator import \
+    StreamSlicerPartitionGenerator, DeclarativePartitionFactory
 from airbyte_cdk.sources.declarative.transformations import (
     AddFields,
     RecordTransformation,
@@ -604,7 +607,9 @@
     WeekClampingStrategy,
     Weekday,
 )
-from airbyte_cdk.sources.streams.concurrent.cursor import ConcurrentCursor, CursorField
+from airbyte_cdk.sources.streams.concurrent.cursor import ConcurrentCursor, CursorField, Cursor, FinalStateCursor
+from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream
+from airbyte_cdk.sources.streams.concurrent.helpers import get_primary_key_from_stream
 from airbyte_cdk.sources.streams.concurrent.state_converters.datetime_stream_state_converter import (
     CustomFormatConcurrentStreamStateConverter,
     DateTimeStreamStateConverter,
@@ -634,7 +639,6 @@ def __init__(
         emit_connector_builder_messages: bool = False,
         disable_retries: bool = False,
         disable_cache: bool = False,
-        disable_resumable_full_refresh: bool = False,
         message_repository: Optional[MessageRepository] = None,
         connector_state_manager: Optional[ConnectorStateManager] = None,
         max_concurrent_async_job_count: Optional[int] = None,
@@ -645,7 +649,6 @@ def __init__(
         self._emit_connector_builder_messages = emit_connector_builder_messages
         self._disable_retries = disable_retries
         self._disable_cache = disable_cache
-        self._disable_resumable_full_refresh = disable_resumable_full_refresh
         self._message_repository = message_repository or InMemoryMessageRepository(
             self._evaluate_log_level(emit_connector_builder_messages)
         )
@@ -2035,15 +2038,6 @@ def create_declarative_stream(
             file_uploader=file_uploader,
             incremental_sync=model.incremental_sync,
         )
-        cursor_field = model.incremental_sync.cursor_field if model.incremental_sync else None
-
-        if model.state_migrations:
-            state_transformations = [
-                self._create_component_from_model(state_migration, config, declarative_stream=model)
-                for state_migration in model.state_migrations
-            ]
-        else:
-            state_transformations = []
 
         schema_loader: Union[
             CompositeSchemaLoader,
@@ -2071,6 +2065,15 @@ def create_declarative_stream(
                 options["name"] = model.name
             schema_loader = DefaultSchemaLoader(config=config, parameters=options)
 
+        cursor_field = model.incremental_sync.cursor_field if model.incremental_sync else None
+
+        if model.state_migrations:
+            state_transformations = [
+                self._create_component_from_model(state_migration, config, declarative_stream=model)
+                for state_migration in model.state_migrations
+            ]
+        else:
+            state_transformations = []
         return DeclarativeStream(
             name=model.name or "",
             primary_key=primary_key,
@@ -2185,28 +2188,6 @@ def _build_incremental_cursor(
             return self._create_component_from_model(model=model.incremental_sync, config=config)  # type: ignore[no-any-return]  # Will be created Cursor as stream_slicer_model is model.incremental_sync
         return None
 
-    def _build_resumable_cursor(
-        self,
-        model: Union[
-            AsyncRetrieverModel,
-            CustomRetrieverModel,
-            SimpleRetrieverModel,
-        ],
-        stream_slicer: Optional[PartitionRouter],
-    ) -> Optional[StreamSlicer]:
-        if hasattr(model, "paginator") and model.paginator and not stream_slicer:
-            # For the regular Full-Refresh streams, we use the high level `ResumableFullRefreshCursor`
-            return ResumableFullRefreshCursor(parameters={})
-        elif stream_slicer:
-            # For the Full-Refresh sub-streams, we use the nested `ChildPartitionResumableFullRefreshCursor`
-            return PerPartitionCursor(
-                cursor_factory=CursorFactory(
-                    create_function=partial(ChildPartitionResumableFullRefreshCursor, {})
-                ),
-                partition_router=stream_slicer,
-            )
-        return None
-
     def _merge_stream_slicers(
         self, model: DeclarativeStreamModel, config: Config
     ) -> Optional[StreamSlicer]:
@@ -2243,11 +2224,7 @@ def _merge_stream_slicers(
         if model.incremental_sync:
             return self._build_incremental_cursor(model, stream_slicer, config)
 
-        return (
-            stream_slicer
-            if self._disable_resumable_full_refresh
-            else self._build_resumable_cursor(retriever_model, stream_slicer)
-        )
+        return stream_slicer
 
     def create_default_error_handler(
         self, model: DefaultErrorHandlerModel, config: Config, **kwargs: Any
@@ -2529,9 +2506,6 @@ def create_schema_type_identifier(
     def create_dynamic_schema_loader(
         self, model: DynamicSchemaLoaderModel, config: Config, **kwargs: Any
     ) -> DynamicSchemaLoader:
-        stream_slicer = self._build_stream_slicer_from_partition_router(model.retriever, config)
-        combined_slicers = self._build_resumable_cursor(model.retriever, stream_slicer)
-
         schema_transformations = []
         if model.schema_transformations:
             for transformation_model in model.schema_transformations:
@@ -2544,7 +2518,7 @@ def create_dynamic_schema_loader(
             config=config,
             name=name,
             primary_key=None,
-            stream_slicer=combined_slicers,
+            stream_slicer=self._build_stream_slicer_from_partition_router(model.retriever, config),
             transformations=[],
             use_cache=True,
             log_formatter=(
@@ -3808,15 +3782,12 @@ def create_components_mapping_definition(
     def create_http_components_resolver(
         self, model: HttpComponentsResolverModel, config: Config, stream_name: Optional[str] = None
     ) -> Any:
-        stream_slicer = self._build_stream_slicer_from_partition_router(model.retriever, config)
-        combined_slicers = self._build_resumable_cursor(model.retriever, stream_slicer)
-
         retriever = self._create_component_from_model(
             model=model.retriever,
             config=config,
             name=f"{stream_name if stream_name else '__http_components_resolver'}",
             primary_key=None,
-            stream_slicer=stream_slicer if stream_slicer else combined_slicers,
+            stream_slicer=self._build_stream_slicer_from_partition_router(model.retriever, config),
             transformations=[],
         )
 

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -1055,152 +1055,6 @@ def test_stream_with_incremental_and_async_retriever_with_partition_router(use_l
     assert stream_slices == expected_stream_slices
 
 
-def test_resumable_full_refresh_stream():
-    content = """
-decoder:
-  type: JsonDecoder
-extractor:
-  type: DpathExtractor
-selector:
-  type: RecordSelector
-  record_filter:
-    type: RecordFilter
-    condition: "{{ record['id'] > stream_state['id'] }}"
-metadata_paginator:
-    type: DefaultPaginator
-    page_size_option:
-      type: RequestOption
-      inject_into: body_json
-      field_path: ["variables", "page_size"]
-    page_token_option:
-      type: RequestPath
-    pagination_strategy:
-      type: "CursorPagination"
-      cursor_value: "{{ response._metadata.next }}"
-      page_size: 10
-requester:
-  type: HttpRequester
-  url_base: "https://api.sendgrid.com/v3/"
-  http_method: "GET"
-  authenticator:
-    type: BearerAuthenticator
-    api_token: "{{ config['apikey'] }}"
-  request_parameters:
-    unit: "day"
-retriever:
-  paginator:
-    type: NoPagination
-  decoder:
-    $ref: "#/decoder"
-partial_stream:
-  type: DeclarativeStream
-  schema_loader:
-    type: JsonFileSchemaLoader
-    file_path: "./source_sendgrid/schemas/{{ parameters.name }}.json"
-list_stream:
-  $ref: "#/partial_stream"
-  $parameters:
-    name: "lists"
-    extractor:
-      $ref: "#/extractor"
-      field_path: ["{{ parameters['name'] }}"]
-  name: "lists"
-  primary_key: "id"
-  retriever:
-    $ref: "#/retriever"
-    requester:
-      $ref: "#/requester"
-      path: "{{ next_page_token['next_page_url'] }}"
-    paginator:
-      $ref: "#/metadata_paginator"
-    record_selector:
-      $ref: "#/selector"
-  transformations:
-    - type: AddFields
-      fields:
-      - path: ["extra"]
-        value: "{{ response.to_add }}"
-check:
-  type: CheckStream
-  stream_names: ["list_stream"]
-spec:
-  type: Spec
-  documentation_url: https://airbyte.com/#yaml-from-manifest
-  connection_specification:
-    title: Test Spec
-    type: object
-    required:
-      - api_key
-    additionalProperties: false
-    properties:
-      api_key:
-        type: string
-        airbyte_secret: true
-        title: API Key
-        description: Test API Key
-        order: 0
-  advanced_auth:
-    auth_flow_type: "oauth2.0"
-    """
-    parsed_manifest = YamlDeclarativeSource._parse(content)
-    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
-    resolved_manifest["type"] = "DeclarativeSource"
-    manifest = transformer.propagate_types_and_parameters("", resolved_manifest, {})
-
-    stream_manifest = manifest["list_stream"]
-    assert stream_manifest["type"] == "DeclarativeStream"
-    stream = factory.create_component(
-        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
-    )
-
-    assert isinstance(stream, DeclarativeStream)
-    assert stream.primary_key == "id"
-    assert stream.name == "lists"
-    assert stream._stream_cursor_field.string == ""
-
-    assert isinstance(stream.retriever, SimpleRetriever)
-    assert stream.retriever.primary_key == stream.primary_key
-    assert stream.retriever.name == stream.name
-
-    assert isinstance(stream.retriever.record_selector, RecordSelector)
-
-    assert isinstance(stream.retriever.stream_slicer, ResumableFullRefreshCursor)
-    assert isinstance(stream.retriever.cursor, ResumableFullRefreshCursor)
-
-    assert isinstance(stream.retriever.paginator, DefaultPaginator)
-    assert isinstance(stream.retriever.paginator.decoder, PaginationDecoderDecorator)
-    for string in stream.retriever.paginator.page_size_option.field_path:
-        assert isinstance(string, InterpolatedString)
-    assert len(stream.retriever.paginator.page_size_option.field_path) == 2
-    assert stream.retriever.paginator.page_size_option.inject_into == RequestOptionType.body_json
-    assert isinstance(stream.retriever.paginator.page_token_option, RequestPath)
-    assert stream.retriever.paginator.url_base.string == "https://api.sendgrid.com/v3/"
-    assert stream.retriever.paginator.url_base.default == "https://api.sendgrid.com/v3/"
-
-    assert isinstance(stream.retriever.paginator.pagination_strategy, CursorPaginationStrategy)
-    assert isinstance(
-        stream.retriever.paginator.pagination_strategy.decoder, PaginationDecoderDecorator
-    )
-    assert (
-        stream.retriever.paginator.pagination_strategy._cursor_value.string
-        == "{{ response._metadata.next }}"
-    )
-    assert (
-        stream.retriever.paginator.pagination_strategy._cursor_value.default
-        == "{{ response._metadata.next }}"
-    )
-    assert stream.retriever.paginator.pagination_strategy.page_size == 10
-
-    checker = factory.create_component(
-        model_type=CheckStreamModel, component_definition=manifest["check"], config=input_config
-    )
-
-    assert isinstance(checker, CheckStream)
-    streams_to_check = checker.stream_names
-    assert len(streams_to_check) == 1
-    assert list(streams_to_check)[0] == "list_stream"
-
-
 def test_incremental_data_feed():
     content = """
 selector:
@@ -2592,7 +2446,7 @@ def test_default_schema_loader(self):
                 "values": "{{config['repos']}}",
                 "cursor_field": "a_key",
             },
-            PerPartitionCursor,
+            ListPartitionRouter,
             id="test_create_simple_retriever_with_partition_router",
         ),
         pytest.param(

unit_tests/sources/declarative/test_manifest_declarative_source.py

@@ -1818,8 +1818,8 @@ def _create_page(response_body):
             [
                 call({}, {}, None),
                 call(
-                    {"next_page_token": "next"},
-                    {"next_page_token": "next"},
+                    {},
+                    {},
                     {"next_page_token": "next"},
                 ),
             ],
@@ -1907,16 +1907,9 @@ def _create_page(response_body):
             ),
             [{"ABC": 0, "partition": 0}, {"AED": 1, "partition": 0}, {"ABC": 2, "partition": 1}],
             [
-                call({"states": []}, {"partition": "0"}, None),
+                call({}, {"partition": "0"}, None),
                 call(
-                    {
-                        "states": [
-                            {
-                                "partition": {"partition": "0"},
-                                "cursor": {"__ab_full_refresh_sync_complete": True},
-                            }
-                        ]
-                    },
+                    {},
                     {"partition": "1"},
                     None,
                 ),
@@ -2022,17 +2015,10 @@ def _create_page(response_body):
                 {"ABC": 2, "partition": 1},
             ],
             [
-                call({"states": []}, {"partition": "0"}, None),
-                call({"states": []}, {"partition": "0"}, {"next_page_token": "next"}),
+                call({}, {"partition": "0"}, None),
+                call({}, {"partition": "0"}, {"next_page_token": "next"}),
                 call(
-                    {
-                        "states": [
-                            {
-                                "partition": {"partition": "0"},
-                                "cursor": {"__ab_full_refresh_sync_complete": True},
-                            }
-                        ]
-                    },
+                    {},
                     {"partition": "1"},
                     None,
                 ),
@@ -2201,12 +2187,12 @@ def test_only_parent_streams_use_cache():
 
     # Parent stream created for substream
     assert (
-        streams[1].retriever.stream_slicer._partition_router.parent_stream_configs[0].stream.name
+        streams[1].retriever.stream_slicer.parent_stream_configs[0].stream.name
         == "applications"
     )
     assert (
         streams[1]
-        .retriever.stream_slicer._partition_router.parent_stream_configs[0]
+        .retriever.stream_slicer.parent_stream_configs[0]
         .stream.retriever.requester.use_cache
     )