Refactor lazy read

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2873,6 +2873,15 @@ definitions:
       type:
         type: string
         enum: [ParentStreamConfig]
+      lazy_read_pointer:
+        title: Lazy Read Pointer
+        description: If set, this will enable lazy reading, using the initial read of parent records to extract child records.
+        type: array
+        default: [ ]
+        items:
+          - type: string
+        interpolation_context:
+          - config
       parent_key:
         title: Parent Key
         description: The primary key of records from the parent stream that will be used during the retrieval of records for the current substream. This parent identifier field is typically a characteristic of the child records being extracted from the source API.
@@ -3242,15 +3251,6 @@ definitions:
           - "$ref": "#/definitions/IterableDecoder"
           - "$ref": "#/definitions/XmlDecoder"
           - "$ref": "#/definitions/ZipfileDecoder"
-      lazy_read_pointer:
-        title: Lazy Read Pointer
-        description: If set, this will enable lazy reading, using the initial read of parent records to extract child records.
-        type: array
-        default: [ ]
-        items:
-          - type: string
-        interpolation_context:
-          - config
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -609,7 +609,9 @@ class OAuthAuthenticator(BaseModel):
     scopes: Optional[List[str]] = Field(
         None,
         description="List of scopes that should be granted to the access token.",
-        examples=[["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"]],
+        examples=[
+            ["crm.list.read", "crm.objects.contacts.read", "crm.schema.contacts.read"]
+        ],
         title="Scopes",
     )
     token_expiry_date: Optional[str] = Field(
@@ -1078,24 +1080,28 @@ class OAuthConfigSpecification(BaseModel):
     class Config:
         extra = Extra.allow
 
-    oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = Field(
-        None,
-        description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['app_id']\n    }\n  }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['info', 'app_id']\n    }\n  }",
-        examples=[
-            {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}},
-            {
-                "app_id": {
-                    "type": "string",
-                    "path_in_connector_config": ["info", "app_id"],
-                }
-            },
-        ],
-        title="OAuth user input",
+    oauth_user_input_from_connector_config_specification: Optional[Dict[str, Any]] = (
+        Field(
+            None,
+            description="OAuth specific blob. This is a Json Schema used to validate Json configurations used as input to OAuth.\nMust be a valid non-nested JSON that refers to properties from ConnectorSpecification.connectionSpecification\nusing special annotation 'path_in_connector_config'.\nThese are input values the user is entering through the UI to authenticate to the connector, that might also shared\nas inputs for syncing data via the connector.\nExamples:\nif no connector values is shared during oauth flow, oauth_user_input_from_connector_config_specification=[]\nif connector values such as 'app_id' inside the top level are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['app_id']\n    }\n  }\nif connector values such as 'info.app_id' nested inside another object are used to generate the API url for the oauth flow,\n  oauth_user_input_from_connector_config_specification={\n    app_id: {\n      type: string\n      path_in_connector_config: ['info', 'app_id']\n    }\n  }",
+            examples=[
+                {"app_id": {"type": "string", "path_in_connector_config": ["app_id"]}},
+                {
+                    "app_id": {
+                        "type": "string",
+                        "path_in_connector_config": ["info", "app_id"],
+                    }
+                },
+            ],
+            title="OAuth user input",
+        )
     )
-    oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = Field(
-        None,
-        description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n  + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n  + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n  + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n  + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n  + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n  - The TikTok Marketing DeclarativeOAuth spec:\n  {\n    "oauth_connector_input_specification": {\n      "type": "object",\n      "additionalProperties": false,\n      "properties": {\n          "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n          "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n          "access_token_params": {\n              "{{ auth_code_key }}": "{{ auth_code_value }}",\n              "{{ client_id_key }}": "{{ client_id_value }}",\n              "{{ client_secret_key }}": "{{ client_secret_value }}"\n          },\n          "access_token_headers": {\n              "Content-Type": "application/json",\n              "Accept": "application/json"\n          },\n          "extract_output": ["data.access_token"],\n          "client_id_key": "app_id",\n          "client_secret_key": "secret",\n          "auth_code_key": "auth_code"\n      }\n    }\n  }',
-        title="DeclarativeOAuth Connector Specification",
+    oauth_connector_input_specification: Optional[OauthConnectorInputSpecification] = (
+        Field(
+            None,
+            description='The DeclarativeOAuth specific blob.\nPertains to the fields defined by the connector relating to the OAuth flow.\n\nInterpolation capabilities:\n- The variables placeholders are declared as `{{my_var}}`.\n- The nested resolution variables like `{{ {{my_nested_var}} }}` is allowed as well.\n\n- The allowed interpolation context is:\n  + base64Encoder - encode to `base64`, {{ {{my_var_a}}:{{my_var_b}} | base64Encoder }}\n  + base64Decorer - decode from `base64` encoded string, {{ {{my_string_variable_or_string_value}} | base64Decoder }}\n  + urlEncoder - encode the input string to URL-like format, {{ https://test.host.com/endpoint | urlEncoder}}\n  + urlDecorer - decode the input url-encoded string into text format, {{ urlDecoder:https%3A%2F%2Fairbyte.io | urlDecoder}}\n  + codeChallengeS256 - get the `codeChallenge` encoded value to provide additional data-provider specific authorisation values, {{ {{state_value}} | codeChallengeS256 }}\n\nExamples:\n  - The TikTok Marketing DeclarativeOAuth spec:\n  {\n    "oauth_connector_input_specification": {\n      "type": "object",\n      "additionalProperties": false,\n      "properties": {\n          "consent_url": "https://ads.tiktok.com/marketing_api/auth?{{client_id_key}}={{client_id_value}}&{{redirect_uri_key}}={{ {{redirect_uri_value}} | urlEncoder}}&{{state_key}}={{state_value}}",\n          "access_token_url": "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/",\n          "access_token_params": {\n              "{{ auth_code_key }}": "{{ auth_code_value }}",\n              "{{ client_id_key }}": "{{ client_id_value }}",\n              "{{ client_secret_key }}": "{{ client_secret_value }}"\n          },\n          "access_token_headers": {\n              "Content-Type": "application/json",\n              "Accept": "application/json"\n          },\n          "extract_output": ["data.access_token"],\n          "client_id_key": "app_id",\n          "client_secret_key": "secret",\n          "auth_code_key": "auth_code"\n      }\n    }\n  }',
+            title="DeclarativeOAuth Connector Specification",
+        )
     )
     complete_oauth_output_specification: Optional[Dict[str, Any]] = Field(
         None,
@@ -1113,7 +1119,9 @@ class Config:
     complete_oauth_server_input_specification: Optional[Dict[str, Any]] = Field(
         None,
         description="OAuth specific blob. This is a Json Schema used to validate Json configurations persisted as Airbyte Server configurations.\nMust be a valid non-nested JSON describing additional fields configured by the Airbyte Instance or Workspace Admins to be used by the\nserver when completing an OAuth flow (typically exchanging an auth code for refresh token).\nExamples:\n    complete_oauth_server_input_specification={\n      client_id: {\n        type: string\n      },\n      client_secret: {\n        type: string\n      }\n    }",
-        examples=[{"client_id": {"type": "string"}, "client_secret": {"type": "string"}}],
+        examples=[
+            {"client_id": {"type": "string"}, "client_secret": {"type": "string"}}
+        ],
         title="OAuth input specification",
     )
     complete_oauth_server_output_specification: Optional[Dict[str, Any]] = Field(
@@ -1766,7 +1774,9 @@ class RecordSelector(BaseModel):
         description="Responsible for filtering records to be emitted by the Source.",
         title="Record Filter",
     )
-    schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field(
+    schema_normalization: Optional[
+        Union[SchemaNormalization, CustomSchemaNormalization]
+    ] = Field(
         SchemaNormalization.None_,
         description="Responsible for normalization according to the schema.",
         title="Schema Normalization",
@@ -1987,7 +1997,9 @@ class Config:
         description="Component used to fetch data incrementally based on a time field in the data.",
         title="Incremental Sync",
     )
-    name: Optional[str] = Field("", description="The stream name.", example=["Users"], title="Name")
+    name: Optional[str] = Field(
+        "", description="The stream name.", example=["Users"], title="Name"
+    )
     primary_key: Optional[PrimaryKey] = Field(
         "", description="The primary key of the stream.", title="Primary Key"
     )
@@ -2205,6 +2217,11 @@ class DynamicSchemaLoader(BaseModel):
 
 class ParentStreamConfig(BaseModel):
     type: Literal["ParentStreamConfig"]
+    lazy_read_pointer: Optional[List[str]] = Field(
+        [],
+        description="If set, this will enable lazy reading, using the initial read of parent records to extract child records.",
+        title="Lazy Read Pointer",
+    )
     parent_key: str = Field(
         ...,
         description="The primary key of records from the parent stream that will be used during the retrieval of records for the current substream. This parent identifier field is typically a characteristic of the child records being extracted from the source API.",
@@ -2240,7 +2257,9 @@ class ParentStreamConfig(BaseModel):
 
 class StateDelegatingStream(BaseModel):
     type: Literal["StateDelegatingStream"]
-    name: str = Field(..., description="The stream name.", example=["Users"], title="Name")
+    name: str = Field(
+        ..., description="The stream name.", example=["Users"], title="Name"
+    )
     full_refresh_stream: DeclarativeStream = Field(
         ...,
         description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.",
@@ -2277,7 +2296,11 @@ class SimpleRetriever(BaseModel):
             CustomPartitionRouter,
             ListPartitionRouter,
             SubstreamPartitionRouter,
-            List[Union[CustomPartitionRouter, ListPartitionRouter, SubstreamPartitionRouter]],
+            List[
+                Union[
+                    CustomPartitionRouter, ListPartitionRouter, SubstreamPartitionRouter
+                ]
+            ],
         ]
     ] = Field(
         [],
@@ -2300,11 +2323,6 @@ class SimpleRetriever(BaseModel):
         description="Component decoding the response so records can be extracted.",
         title="Decoder",
     )
-    lazy_read_pointer: Optional[List[str]] = Field(
-        [],
-        description="If set, this will enable lazy reading, using the initial read of parent records to extract child records.",
-        title="Lazy Read Pointer",
-    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -2326,7 +2344,9 @@ class AsyncRetriever(BaseModel):
     )
     download_extractor: Optional[
         Union[CustomRecordExtractor, DpathExtractor, ResponseToFileExtractor]
-    ] = Field(None, description="Responsible for fetching the records from provided urls.")
+    ] = Field(
+        None, description="Responsible for fetching the records from provided urls."
+    )
     creation_requester: Union[CustomRequester, HttpRequester] = Field(
         ...,
         description="Requester component that describes how to prepare HTTP requests to send to the source API to create the async server-side job.",
@@ -2360,7 +2380,11 @@ class AsyncRetriever(BaseModel):
             CustomPartitionRouter,
             ListPartitionRouter,
             SubstreamPartitionRouter,
-            List[Union[CustomPartitionRouter, ListPartitionRouter, SubstreamPartitionRouter]],
+            List[
+                Union[
+                    CustomPartitionRouter, ListPartitionRouter, SubstreamPartitionRouter
+                ]
+            ],
         ]
     ] = Field(
         [],
@@ -2428,10 +2452,12 @@ class DynamicDeclarativeStream(BaseModel):
     stream_template: DeclarativeStream = Field(
         ..., description="Reference to the stream template.", title="Stream Template"
     )
-    components_resolver: Union[HttpComponentsResolver, ConfigComponentsResolver] = Field(
-        ...,
-        description="Component resolve and populates stream templates with components values.",
-        title="Components Resolver",
+    components_resolver: Union[HttpComponentsResolver, ConfigComponentsResolver] = (
+        Field(
+            ...,
+            description="Component resolve and populates stream templates with components values.",
+            title="Components Resolver",
+        )
     )
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2528,6 +2528,10 @@ def create_parent_stream_config(
             if model.request_option
             else None
         )
+
+        if "*" in model.lazy_read_pointer:
+            raise ValueError("The '*' wildcard in 'lazy_read_pointer' is not supported — only direct paths are allowed.")
+
         return ParentStreamConfig(
             parent_key=model.parent_key,
             request_option=request_option,
@@ -2537,6 +2541,7 @@ def create_parent_stream_config(
             incremental_dependency=model.incremental_dependency or False,
             parameters=model.parameters or {},
             extra_fields=model.extra_fields,
+            lazy_read_pointer=model.lazy_read_pointer
         )
 
     @staticmethod
@@ -2740,42 +2745,13 @@ def create_simple_retriever(
             model.ignore_stream_slicer_parameters_on_paginated_requests or False
         )
 
-        if model.lazy_read_pointer and not hasattr(model, "partition_router"):
-            raise ValueError(
-                "LazySimpleRetriever requires a 'partition_router' when 'lazy_read_pointer' is set. "
-                "Please either define 'partition_router' or remove 'lazy_read_pointer' from the model."
-            )
+        if hasattr(model, "partition_router") and model.partition_router and model.partition_router.type == "SubstreamPartitionRouter" and not bool(
+            self._connector_state_manager.get_stream_state(name, None)) and any(parent_stream_config.lazy_read_pointer for parent_stream_config in model.partition_router.parent_stream_configs):
 
-        if model.lazy_read_pointer and not bool(
-            self._connector_state_manager.get_stream_state(name, None)
-        ):
-            if model.partition_router.type != "SubstreamPartitionRouter":  # type: ignore[union-attr] # model.partition_router has BaseModel type
+            if incremental_sync.step or incremental_sync.cursor_granularity:
                 raise ValueError(
-                    "LazySimpleRetriever only supports 'SubstreamPartitionRouterModel' as the 'partition_router' type. "  # type: ignore[union-attr] # model.partition_router has BaseModel type
-                    f"Found: '{model.partition_router.type}'."
+                    f"Found more that one slice per parent. LazySimpleRetriever only supports single slice read for stream - {name}."
                 )
-            lazy_read_pointer = []
-            for i, path in enumerate(model.lazy_read_pointer):
-                if path == "*":
-                    raise ValueError(
-                        f"'lazy_read_pointer' support only direct pointing. Found: '* as a {i} element in the pointer.'"
-                    )
-
-                lazy_read_pointer.append(InterpolatedString.create(path, parameters=model.parameters or {}))
-
-            lazy_read_pointer = [
-                InterpolatedString.create(path, parameters=model.parameters or {})
-                for path in model.lazy_read_pointer
-            ]
-            partition_router = self._create_component_from_model(
-                model=model.partition_router,  # type: ignore[arg-type] # model.partition_router has BaseModel type
-                config=config,  # type: ignore[arg-type]
-            )
-            stream_slicer = (
-                self._create_component_from_model(model=incremental_sync, config=config)
-                if incremental_sync
-                else SinglePartitionRouter(parameters={})
-            )
 
             return LazySimpleRetriever(
                 name=name,
@@ -2789,8 +2765,6 @@ def create_simple_retriever(
                 config=config,
                 ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
                 parameters=model.parameters or {},
-                partition_router=partition_router,
-                lazy_read_pointer=lazy_read_pointer,
             )
 
         if self._limit_slices_fetched or self._emit_connector_builder_messages: