spike source emit request and response

[girarda]

What

The connector builder requires access to the requests and responses that produced a given set of records.
This functionality is not currently implemented in the CDK.
Its implementation requires modifying the CDK at multiple layer because the Source and Stream classes are not aware of HTTP requests and responses.

How

Intercepting the requests and responses

The HttpStream class already logs the HTTP requests sent to the API as well as the HTTP responses we receive.
Instead of directly logging the requests and responses, the class can implement 2 new methods

1. log_request
2. log_response
   By default, these methods will log the requests/responses like the HttpStream currently does.

Exposing the requests/responses to the Source

The SimpleRetriever class can then overwrite these methods to keep track of them in memory:

def log_request(self, request: requests.PreparedRequest):
    self._request = {"url": request.url, "body": request.body, "headers": dict(request.headers)}

Before returning the parsed records from parse_response, the SimpleRetriever can amend the records with metadata representing the request and response that produced the record

for r in records:
    r["_ab_response"] = response.content
    r["_ab_request"] = self._request
return records

Exposing the requests/responses as AirbyteMessages

Since the AirbyteRecords produced by the AbstractSource now contain metadata, the DeclarativeSource can extract the metadata from the records and emit the requests and responses as their own AirbyteMessage.

def read(
    self,
    logger: logging.Logger,
    config: Mapping[str, Any],
    catalog: ConfiguredAirbyteCatalog,
    state: Union[List[AirbyteStateMessage], MutableMapping[str, Any]] = None,
) -> Iterator[AirbyteMessage]:
    last_request = None
    for airbyte_message in super().read(logger, config, catalog, state):
        if airbyte_message.record and airbyte_message.record.data["_ab_request"] != last_request:
            yield AirbyteMessage(
                type=MessageType.LOG, log=AirbyteLogMessage(message=str(airbyte_message.record.data["_ab_request"]), level=Level.INFO)
            )
            yield AirbyteMessage(
                type=MessageType.LOG, log=AirbyteLogMessage(message=str(airbyte_message.record.data["_ab_response"]), level=Level.INFO)
            )
        last_request = airbyte_message.record.data.pop("_ab_request")
        airbyte_message.record.data.pop("_ab_response")
        yield airbyte_message

Exposing the requests/responses to the UI

The backend server that will run the DeclarativeSource can then extract the AirbyteRecords and group them as needed for the connector builder.

Additional decisions to make

• What type of AirbyteMessage should contain the request/response?
• Should there be a single metadata field added to the record's data mapping or one for request and one for response?
• Are there any concerns with the SimpleRetriever amending the data produced? It tightly couples it with the DeclarativeSource, but I don't see them being used separately anyway
• Should the additional logic live in the existing classes, or should we create wrapper classes?

[girarda]

complexity: also need to check if we're processing a new stream slice