Connector/Source: Publish the *request* object in the record transformation like the config, stream_slice, stream_interval, etc. are

[rpopov]

Status in Airbyte 1.2.0
The record transformation allows removing existing fields from the record and adding new fields by calculating them using:

• JINJA templates
• pre-defined objects found in the transformation's context:
  • record
  • stream_state
  • stream_slice
  • stream_interval
  • stream_partition
• the access to the response raw data is redirected through the record object, which by definition is only part of the response. Converting the response into valuable records sometimes it needs data beyond the scope of the record object, which is not accessible.

Example
The JIRA /issue response:

{
  "expand": "renderedFields,names,schema,operations,editmeta,changelog,versionedRepresentations",
  "id": "2759961",
  "self": "https://jira-test.paysafe.cloud/rest/api/2/issue/2759961",
  "key": "EF-3131",
  "fields": {
    "customfield_19440": {
      "self": "https://jira-test.paysafe.cloud/rest/api/2/customFieldOption/23882",
      "value": "Perú",
      "id": "23882",
      "disabled": false
    },
...
    "customfield_12073": "EF-3130"
  },
  "names": {
    "customfield_19442": "End Time",
...
    "customfield_17709": "BU Legal Lead"
  },
  "schema": {
    "customfield_19442": {
      "type": "date",
      "custom": "com.atlassian.jira.plugin.system.customfieldtypes:datepicker",
      "customId": 19442
    },
...
    "customfield_17709": {
      "type": "string",
      "custom": "com.atlassian.jira.plugin.system.customfieldtypes:textfield",
      "customId": 17709
    }
  },
}
]

In JIRA the list of custom fields is highly dynamic, and it makes no practical sense to have a single record per issue, combining in it all standard and custom fields.
Instead, the JIRA issue can be represented using 2 tables: ISSUE 1 --- * CUSTOM_FIELD in one-to-many / master-detail relation.
Turning the response into a list of records, one per custom field, could be done by iterating over the schema.* sub-list in the response, but taking their values from the fields map and taking their human-readble names from the name map.

Problem
In this configuration, the fields and names maps are not accessible. They could be, if the response object were available in the Tranformations' context. The response object exists in the context of the Pagination section.

Suggestion

• Publish the response object in the context of the Transformations section/phase, as available in the Pagination section.

[marcosmarxm]

Do you want to have a stream that translates the custom objects into a single table? (to me still kind confuse the request)

[rpopov]

Yes, I want to convert the objects held as values in the schema map from the response into separate records in a detail DB table (e.g. CUSTOM_FIELD_VALUE). These records will be related to a single record for the issue in a master DB table (e.g., ISSUE) that is already retrieved.

I realized that the path schema.* allows Airbyte build a list of the values out of the JSON response (see above). The list looks like:

[
  {
    "type": "date",
    "custom": "com.atlassian.jira.plugin.system.customfieldtypes:datepicker",
    "customId": 19442,
    "issue_key": "COWAL-172",
    "custom_field_id": 19442
  },
  {
    "type": "string",
    "custom": "com.atlassian.jira.plugin.system.customfieldtypes:textarea",
    "customId": 19684,
    "issue_key": "COWAL-172",
    "custom_field_id": 19684
  },
...
]

This list gives the type, ID of the custom field. With some tricks, it even calculated the issue key (to become the foreign key in CUSTOM_FIELD_VALUE table to the ISSUE table), but:

• it lacks the field's name, which could be taken from the sibling map names,
• it lacks the field's value, which could be taken from the sibling map fields.

Unfortunately:

• the record selection path schema.* hides those sibling maps (objects);
• selecting and iterating over the values in the fields and names collections just fail, and schema.* is the only one working.

If we had the response object in the record transformation's context (like the config, stream_slice, stream_state objects) so an expression like:

• {{ response.fields.get("customfield_"+record.custom_field_id) }} could calculate the field value
• {{ response.names.get("customfield_"+record.custom_field_id) }} could calculate the field's human-readable name.

Note: It would make sense to have the response object also in the context of the record selector too.

The trace to the record transformation's context could be found as:

Traceback (most recent call last):
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/abstract_source.py", line 133, in read
    yield from self._read_stream(
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/abstract_source.py", line 239, in _read_stream
    for record_data_or_message in record_iterator:
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/streams/core.py", line 192, in read
    for record_data_or_message in records:
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/declarative_stream.py", line 137, in read_records
    yield from self.retriever.read_records(self.get_json_schema(), stream_slice)  # type: ignore # records are of the correct type
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py", line 378, in read_records
    for stream_data in self._read_pages(record_generator, self.state, _slice):
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py", line 302, in _read_pages
    yield from records_generator_fn(response)
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py", line 448, in _parse_records
    yield from self._parse_response(
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py", line 248, in _parse_response
    for record in record_generator:
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/extractors/record_selector.py", line 64, in select_records
    yield from self.filter_and_transform(all_data, stream_state, records_schema, stream_slice, next_page_token)
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/extractors/record_selector.py", line 85, in filter_and_transform
    for data in normalized_data:
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/extractors/record_selector.py", line 93, in _normalize_by_schema
    for record in records:
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/extractors/record_selector.py", line 122, in _transform
    transformation.transform(record, config=self.config, stream_state=stream_state, stream_slice=stream_slice)  # type: ignore  # record has type Mapping[str, Any], but Dict[str, Any] expected
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/transformations/add_fields.py", line 124, in transform
    value = parsed_field.value.eval(config, valid_types=valid_types, **kwargs)
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/interpolation/interpolated_string.py", line 48, in eval
    evaluated = self._interpolation.eval(self.string, config, self.default, parameters=self._parameters, **kwargs)
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/interpolation/jinja.py", line 97, in eval
    result = self._eval(input_str, context)
  File "/home/airbyte/.pyenv/versions/3.10.14/lib/python3.10/site-packages/airbyte_cdk/sources/declarative/interpolation/jinja.py", line 122, in _eval
    raise ValueError(f"Jinja macro has undeclared variables: {undeclared_not_in_context}. Context: {context}")

Whis is reported when a field transformation includes {{response}}.

The transformation's context already lists:

ValueError: 
  Jinja macro has undeclared variables: {'response'}. 
  Context: 
  {
      'config': {....},
      'parameters': {},
      'record': {
          'type': 'date',
          'custom': 'com.atlassian.jira.plugin.system.customfieldtypes:datepicker',
          'customId': 19442,
          'issue_key': 'EF-3131',
          'custom_field_id': 19442
      },
      'stream_state': {
          'states': []
      },
      'stream_slice': {
          'issue_key': 'EF-3131',
          'parent_slice': {}
      },
      'stream_interval': {
          'issue_key': 'EF-3131',
          'parent_slice': {}
      },
      'stream_partition': {
          'issue_key': 'EF-3131',
          'parent_slice': {}
      }
  }

[rpopov]

An alternative could be providing the (current) response object bound in the stream_state object that is already in the context.
Example:

Context: {
...
      'stream_state': {
          'states': []
      }
...
}

becomes:

Context: {
...
      'stream_state': {
          'states': [],
          'response': {...}
      }
...
}

This will keep the existing public interfaces and contracts unchanged. The only change would be in the contents of the stream_state object. The existing implementation would publish the response object for filtering and transformation steps (see record_selector.py).
This could become a pattern for adding more objects in the context in the future.

[rpopov]

Analyzing in much more detail the design of the record extractors and the RecordSelector class, where the processing happens, and the options not to change the published interfaces and contracts, I see:

• The only case where the response is not a list of records is the case of JSON/XML/other response containing nested records:
  • This is the case where having access to the whole response is beneficial;
  • In this case, the direct-path extractor (DpathExtractor) is used to navigate to the records.
• The place where the response is parsed before it is split into records is in the direct-path extractor:
  • there is no access to the transformation's context;
  • it has no access to the stream_* objects;
  • there is no way to make the stream_* objects accessible without vast interface changes.
• In responses that are lists of records (JSONL, files, etc.), the processing of the individual record
  • may use direct-path extractor to retrieve nested parts of each record;
  • cannot benefit from having the whole list of records;
  • but could benefit from accessing the whole record itself.
    Then, the response should be bound somewhere else.

Suggestion

• Bind the response root object in the record itself to a specific key.
• The key should indicate it is for service purposes (maybe $response, $root, like the $parameters and $ref keys in the schema)
• The key should not be stored - it should be removed explicitly as part of/after the record transformation or it should be ignored.

[rpopov]

airbytehq/airbyte-python-cdk#193