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

[ChristoGrab]

WIP

What

To support GraphQL queries natively in the CDK, we need to add handling of nested structures when injecting a RequestOption into a request body (specifically with body_json injection).

How

• The RequestOption class now includes an optional field_path param, and each instance must declare a field_path OR field_name.
• The new inject_into_request method determines how to handle the injection based on the existense of field_path and the specific RequestOptionType. As stated above, only body_json injections will support nested fields.
• The combine_mappings utility function now handles nested structures when checking for mapping conflicts.
• The following components now call on inject_into_request rather than handling the logic themselves:
  1. ApiKeyAuthenticator
  2. DatetimeBasedCursor

Suggested review order

1. request_option.py: Has the main updated logic for validation and the new method inject_into_request
2. mapping_helpers.py: Added a couple helper methods and extended the main function for catching conflicts in nested structures.
3. token.py and datetime_based_cursor.py: Small updates to leverage the new injection logic.
4. declarative_component_schema.yaml and declarative_component_schema.py: Updated schema + model
5. Everything else is unit testing

TODO

Update the remaining components that allow request_option injection to leverage the new logic. The list is:

1. ListPartitionRouter
2. SubstreamPartitionRouter
3. HttpRequester
4. DefaultPaginator
5. DatetimeBasedRequestOptionsProvider
6. SimpleRetriever

[maxi297]

Adding some comments on this PR even if it is draft. Poke me when you want me to check it again

[maxi297]

Should we slowly deprecate this field in favor of field_path and therefore add this information in the description/title?

[maxi297]

I think this is overly cautious and wouldn't validate that. In Python, any parameter can be of any type but we don't validate them everywhere. We have mypy for that which should hopefully catch most of these. I think at some point, there are readability concerns that we need to balance. Was there a specific gap we wanted to address?

The some comment applies to the self.field_path condition below.

[maxi297]

nit: since only field_name or field_path will be defined, we can probably use elif as elif self.field_path is not None: instead of if self.field_path is not None: to be more explicit

[maxi297]

Could we simplify the logic by having in post_init:

        if self.field_name is not None:
            self.field_path = [InterpolatedString.create(self.field_name, parameters=parameters)]

This way, we would only have one logic to maintain and it would be the field_path one.

[maxi297]

I would precise that it is "If there only one mapping that is a string, it will return this mapping regardless of the other mappings".

That being said, I don't understand this behavior though. Maybe I'm wrong but before, I think it would fail if there was a string and a mapping, right?

[maxi297]

It feels like the following logic is a bit complex. Would there be a way to simplify this? Right now, I see that we:

• Extract all paths
• Compare all paths to find conflicts
• Create a new mapping that we can return

Would it be simpler to do just the last one and validate for conflicts while we do that? Something like:

def combine_mappings(
    mappings: List[Optional[Union[Mapping[str, Any], str]]],
) -> Union[Mapping[str, Any], str]:
    """
    Combine multiple mappings into a single mapping, supporting nested structures.
    If any of the mappings are a string, return that string. Raise errors in the following cases:
    * If there are conflicting paths across mappings (including nested conflicts)
    * If there are multiple string mappings
    * If there are multiple mappings containing keys and one of them is a string
    """
    if not mappings:
        return {}

    # Count how many string options we have, ignoring None values
    string_options = sum(isinstance(mapping, str) for mapping in mappings if mapping is not None)
    if string_options > 1:
        raise ValueError("Cannot combine multiple string options")

    # Filter out None values and empty mappings
    non_empty_mappings = [
        m for m in mappings if m is not None and not (isinstance(m, Mapping) and not m)
    ]

    # If there is only one string option, return it
    if string_options == 1:
        if len(non_empty_mappings) > 1:
            raise ValueError("Cannot combine multiple options if one is a string")
        return next(m for m in non_empty_mappings if isinstance(m, str))

    # Convert all mappings to flat (path, value) pairs for conflict detection
    for other in mappings[1:]:
        if other:
            merge(mappings[0], other)

    return mappings[0]


def merge(a: dict, b: dict, path=[]):
    """
    Blindly and shamelessly taken from https://stackoverflow.com/a/7205107
    """
    for key in b:
        if key in a:
            if isinstance(a[key], dict) and isinstance(b[key], dict):
                merge(a[key], b[key], path + [str(key)])
            elif a[key] != b[key]:
                raise ValueError('Duplicate keys')
        else:
            a[key] = b[key]
    return a

If we are afraid of modifying the mappings in memory, we can create a deepcopy of it as well.