feat: enable handling of nested fields when injecting request_option in request body_json

airbyte_cdk/sources/declarative/auth/token.py

@@ -5,7 +5,7 @@
 import base64
 import logging
 from dataclasses import InitVar, dataclass
-from typing import Any, Mapping, Union
+from typing import Any, Mapping, MutableMapping, Union
 
 import requests
 from cachetools import TTLCache, cached
@@ -45,11 +45,6 @@ class ApiKeyAuthenticator(DeclarativeAuthenticator):
     config: Config
     parameters: InitVar[Mapping[str, Any]]
 
-    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        self._field_name = InterpolatedString.create(
-            self.request_option.field_name, parameters=parameters
-        )
-
     @property
     def auth_header(self) -> str:
         options = self._get_request_options(RequestOptionType.header)
@@ -60,9 +55,9 @@ def token(self) -> str:
         return self.token_provider.get_token()
 
     def _get_request_options(self, option_type: RequestOptionType) -> Mapping[str, Any]:
-        options = {}
+        options: MutableMapping[str, Any] = {}
         if self.request_option.inject_into == option_type:
-            options[self._field_name.eval(self.config)] = self.token
+            self.request_option.inject_into_request(options, self.token, self.config)
         return options
 
     def get_request_params(self) -> Mapping[str, Any]:

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2798,25 +2798,35 @@ definitions:
         enum: [RequestPath]
   RequestOption:
     title: Request Option
-    description: Specifies the key field and where in the request a component's value should be injected.
+    description: Specifies the key field or path and where in the request a component's value should be injected.
     type: object
     required:
       - type
-      - field_name
       - inject_into
     properties:
       type:
         type: string
         enum: [RequestOption]
       field_name:
-        title: Request Option
-        description: Configures which key should be used in the location that the descriptor is being injected into
+        title: Field Name
+        description: Configures which key should be used in the location that the descriptor is being injected into. We hope to eventually deprecate this field in favor of `field_path` for all request_options, but must currently maintain it for backwards compatibility in the Builder.
         type: string
         examples:
           - segment_id
         interpolation_context:
           - config
           - parameters
+      field_path:
+        title: Field Path
+        description: Configures a path to be used for nested structures in JSON body requests (e.g. GraphQL queries)
+        type: array
+        items:
+          type: string
+        examples:
+          - ["data", "viewer", "id"]
+        interpolation_context:
+          - config
+          - parameters
       inject_into:
         title: Inject Into
         description: Configures where the descriptor should be set on the HTTP requests. Note that request parameters that are already encoded in the URL path will not be duplicated.

airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -365,14 +365,15 @@ def _get_request_options(
         options: MutableMapping[str, Any] = {}
         if not stream_slice:
             return options
+
         if self.start_time_option and self.start_time_option.inject_into == option_type:
-            options[self.start_time_option.field_name.eval(config=self.config)] = stream_slice.get(  # type: ignore # field_name is always casted to an interpolated string
-                self._partition_field_start.eval(self.config)
-            )
+            start_time_value = stream_slice.get(self._partition_field_start.eval(self.config))
+            self.start_time_option.inject_into_request(options, start_time_value, self.config)
+
         if self.end_time_option and self.end_time_option.inject_into == option_type:
-            options[self.end_time_option.field_name.eval(config=self.config)] = stream_slice.get(  # type: ignore [union-attr]
-                self._partition_field_end.eval(self.config)
-            )
+            end_time_value = stream_slice.get(self._partition_field_end.eval(self.config))
+            self.end_time_option.inject_into_request(options, end_time_value, self.config)
+
         return options
 
     def should_be_synced(self, record: Record) -> bool:

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -719,7 +719,7 @@ class HttpResponseFilter(BaseModel):
 class TypesMap(BaseModel):
     target_type: Union[str, List[str]]
     current_type: Union[str, List[str]]
-    condition: Optional[str]
+    condition: Optional[str] = None
 
 
 class SchemaTypeIdentifier(BaseModel):
@@ -797,14 +797,11 @@ class DpathFlattenFields(BaseModel):
     field_path: List[str] = Field(
         ...,
         description="A path to field that needs to be flattened.",
-        examples=[
-            ["data"],
-            ["data", "*", "field"],
-        ],
+        examples=[["data"], ["data", "*", "field"]],
         title="Field Path",
     )
     delete_origin_value: Optional[bool] = Field(
-        False,
+        None,
         description="Whether to delete the origin value or keep it. Default is False.",
         title="Delete Origin Value",
     )
@@ -1173,11 +1170,17 @@ class InjectInto(Enum):
 
 class RequestOption(BaseModel):
     type: Literal["RequestOption"]
-    field_name: str = Field(
-        ...,
-        description="Configures which key should be used in the location that the descriptor is being injected into",
+    field_name: Optional[str] = Field(
+        None,
+        description="Configures which key should be used in the location that the descriptor is being injected into. We hope to eventually deprecate this field in favor of `field_path` for all request_options, but must currently maintain it for backwards compatibility in the Builder.",
         examples=["segment_id"],
-        title="Request Option",
+        title="Field Name",
+    )
+    field_path: Optional[List[str]] = Field(
+        None,
+        description="Configures a path to be used for nested structures in JSON body requests (e.g. GraphQL queries)",
+        examples=[["data", "viewer", "id"]],
+        title="Field Path",
     )
     inject_into: InjectInto = Field(
         ...,

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2072,7 +2072,11 @@ def create_request_option(
         model: RequestOptionModel, config: Config, **kwargs: Any
     ) -> RequestOption:
         inject_into = RequestOptionType(model.inject_into.value)
-        return RequestOption(field_name=model.field_name, inject_into=inject_into, parameters={})
+        field_path: Optional[List[Union[InterpolatedString, str]]] = model.field_path  # type: ignore
+        field_name = model.field_name if model.field_name else None
+        return RequestOption(
+            field_name=field_name, field_path=field_path, inject_into=inject_into, parameters={}
+        )
 
     def create_record_selector(
         self,

airbyte_cdk/sources/declarative/partition_routers/list_partition_router.py

@@ -3,7 +3,7 @@
 #
 
 from dataclasses import InitVar, dataclass
-from typing import Any, Iterable, List, Mapping, Optional, Union
+from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Union
 
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.partition_routers.partition_router import PartitionRouter
@@ -100,7 +100,9 @@ def _get_request_option(
         ):
             slice_value = stream_slice.get(self._cursor_field.eval(self.config))
             if slice_value:
-                return {self.request_option.field_name.eval(self.config): slice_value}  # type: ignore # field_name is always casted to InterpolatedString
+                options: MutableMapping[str, Any] = {}
+                self.request_option.inject_into_request(options, slice_value, self.config)
+                return options
             else:
                 return {}
         else:

airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

@@ -4,7 +4,7 @@
 import copy
 import logging
 from dataclasses import InitVar, dataclass
-from typing import TYPE_CHECKING, Any, Iterable, List, Mapping, Optional, Union
+from typing import TYPE_CHECKING, Any, Iterable, List, Mapping, MutableMapping, Optional, Union
 
 import dpath
 
@@ -118,7 +118,7 @@ def get_request_body_json(
     def _get_request_option(
         self, option_type: RequestOptionType, stream_slice: Optional[StreamSlice]
     ) -> Mapping[str, Any]:
-        params = {}
+        params: MutableMapping[str, Any] = {}
         if stream_slice:
             for parent_config in self.parent_stream_configs:
                 if (
@@ -128,13 +128,7 @@ def _get_request_option(
                     key = parent_config.partition_field.eval(self.config)  # type: ignore # partition_field is always casted to an interpolated string
                     value = stream_slice.get(key)
                     if value:
-                        params.update(
-                            {
-                                parent_config.request_option.field_name.eval(  # type: ignore [union-attr]
-                                    config=self.config
-                                ): value
-                            }
-                        )
+                        parent_config.request_option.inject_into_request(params, value, self.config)
         return params
 
     def stream_slices(self) -> Iterable[StreamSlice]:

airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -199,6 +199,9 @@ def _get_request_options(
         Raise a ValueError if there's a key collision
         Returned merged mapping otherwise
         """
+
+        is_body_json = requester_method.__name__ == "get_request_body_json"
+
         return combine_mappings(
             [
                 requester_method(
@@ -208,7 +211,8 @@ def _get_request_options(
                 ),
                 auth_options_method(),
                 extra_options,
-            ]
+            ],
+            allow_same_value_merge=is_body_json,
         )
 
     def _request_headers(

airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py

@@ -187,7 +187,7 @@ def get_request_body_json(
     def _get_request_options(
         self, option_type: RequestOptionType, next_page_token: Optional[Mapping[str, Any]]
     ) -> MutableMapping[str, Any]:
-        options = {}
+        options: MutableMapping[str, Any] = {}
 
         token = next_page_token.get("next_page_token") if next_page_token else None
         if (
@@ -196,15 +196,16 @@ def _get_request_options(
             and isinstance(self.page_token_option, RequestOption)
             and self.page_token_option.inject_into == option_type
         ):
-            options[self.page_token_option.field_name.eval(config=self.config)] = token  # type: ignore # field_name is always cast to an interpolated string
+            self.page_token_option.inject_into_request(options, token, self.config)
+
         if (
             self.page_size_option
             and self.pagination_strategy.get_page_size()
             and self.page_size_option.inject_into == option_type
         ):
-            options[self.page_size_option.field_name.eval(config=self.config)] = (  # type: ignore [union-attr]
-                self.pagination_strategy.get_page_size()
-            )  # type: ignore # field_name is always cast to an interpolated string
+            page_size = self.pagination_strategy.get_page_size()
+            self.page_size_option.inject_into_request(options, page_size, self.config)
+
         return options
 
 

airbyte_cdk/sources/declarative/requesters/request_option.py

@@ -4,9 +4,10 @@
 
 from dataclasses import InitVar, dataclass
 from enum import Enum
-from typing import Any, Mapping, Union
+from typing import Any, List, Literal, Mapping, MutableMapping, Optional, Union
 
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.types import Config
 
 
 class RequestOptionType(Enum):
@@ -26,13 +27,91 @@ class RequestOption:
     Describes an option to set on a request
 
     Attributes:
-        field_name (str): Describes the name of the parameter to inject
+        field_name (str): Describes the name of the parameter to inject. Mutually exclusive with field_path.
+        field_path (list(str)): Describes the path to a nested field as a list of field names.
+          Only valid for body_json injection type, and mutually exclusive with field_name.
         inject_into (RequestOptionType): Describes where in the HTTP request to inject the parameter
     """
 
-    field_name: Union[InterpolatedString, str]
     inject_into: RequestOptionType
     parameters: InitVar[Mapping[str, Any]]
+    field_name: Optional[Union[InterpolatedString, str]] = None
+    field_path: Optional[List[Union[InterpolatedString, str]]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        self.field_name = InterpolatedString.create(self.field_name, parameters=parameters)
+        # Validate inputs. We should expect either field_name or field_path, but not both
+        if self.field_name is None and self.field_path is None:
+            raise ValueError("RequestOption requires either a field_name or field_path")
+
+        if self.field_name is not None and self.field_path is not None:
+            raise ValueError(
+                "Only one of field_name or field_path can be provided to RequestOption"
+            )
+
+        # Nested field injection is only supported for body JSON injection
+        if self.field_path is not None and self.inject_into != RequestOptionType.body_json:
+            raise ValueError(
+                "Nested field injection is only supported for body JSON injection. Please use a top-level field_name for other injection types."
+            )
+
+        # Convert field_name and field_path into InterpolatedString objects if they are strings
+        if self.field_name is not None:
+            self.field_name = InterpolatedString.create(self.field_name, parameters=parameters)
+        elif self.field_path is not None:
+            self.field_path = [
+                InterpolatedString.create(segment, parameters=parameters)
+                for segment in self.field_path
+            ]
+
+    @property
+    def _is_field_path(self) -> bool:
+        """Returns whether this option is a field path (ie, a nested field)"""
+        return self.field_path is not None
+
+    def inject_into_request(
+        self,
+        target: MutableMapping[str, Any],
+        value: Any,
+        config: Config,
+    ) -> None:
+        """
+        Inject a request option value into a target request structure using either field_name or field_path.
+        For non-body-json injection, only top-level field names are supported.
+        For body-json injection, both field names and nested field paths are supported.
+
+        Args:
+            target: The request structure to inject the value into
+            value: The value to inject
+            config: The config object to use for interpolation
+        """
+        if self._is_field_path:
+            if self.inject_into != RequestOptionType.body_json:
+                raise ValueError(
+                    "Nested field injection is only supported for body JSON injection. Please use a top-level field_name for other injection types."
+                )
+
+            assert self.field_path is not None  # for type checker
+            current = target
+            # Convert path segments into strings, evaluating any interpolated segments
+            # Example: ["data", "{{ config[user_type] }}", "id"] -> ["data", "admin", "id"]
+            *path_parts, final_key = [
+                str(
+                    segment.eval(config=config)
+                    if isinstance(segment, InterpolatedString)
+                    else segment
+                )
+                for segment in self.field_path
+            ]
+
+            # Build a nested dictionary structure and set the final value at the deepest level
+            for part in path_parts:
+                current = current.setdefault(part, {})
+            current[final_key] = value
+        else:
+            # For non-nested fields, evaluate the field name if it's an interpolated string
+            key = (
+                self.field_name.eval(config=config)
+                if isinstance(self.field_name, InterpolatedString)
+                else self.field_name
+            )
+            target[str(key)] = value

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

@@ -80,12 +80,13 @@ def _get_request_options(
         options: MutableMapping[str, Any] = {}
         if not stream_slice:
             return options
+
         if self.start_time_option and self.start_time_option.inject_into == option_type:
-            options[self.start_time_option.field_name.eval(config=self.config)] = stream_slice.get(  # type: ignore # field_name is always casted to an interpolated string
-                self._partition_field_start.eval(self.config)
-            )
+            start_time_value = stream_slice.get(self._partition_field_start.eval(self.config))
+            self.start_time_option.inject_into_request(options, start_time_value, self.config)
+
         if self.end_time_option and self.end_time_option.inject_into == option_type:
-            options[self.end_time_option.field_name.eval(config=self.config)] = stream_slice.get(  # type: ignore [union-attr]
-                self._partition_field_end.eval(self.config)
-            )
+            end_time_value = stream_slice.get(self._partition_field_end.eval(self.config))
+            self.end_time_option.inject_into_request(options, end_time_value, self.config)
+
         return options