Stream returns AirbyteMessages #18572

girarda · 2022-10-27T20:21:41Z

What

This is PR 1/2 for emitting requests and responses as AirbyteMessages.
This PR is scoped to refactoring the Stream and the Source such that the Stream returns AirbyteMessage instead of Mapping.
This follow up PR updates the retriever and the HttpStream so requests and responses are emitted as AirbyteLogMessages

Update the Stream interface to yield AirbyteMessage
This allows streams to yield TraceMessages natively

How

Move conversion logic from Mapping[str, Any] -> AirbyteRecord to a reusable function
Add read_records_as_messages to the Stream inteface
read_records_as_messages simply calls read_records` and pipes them to the reusable function
Update the AbstractSource to call read_records_as_messages instead of read_records
read_records isn't modified so child classes can still overwrite is when needed

Testing

In addition to unit tests, I ran the SATs for sendgrid (with a custom docker image to import the CDK changes)

🚨 User Impact 🚨

Are there any breaking changes? What is the end result perceived by the user? If yes, please merge this PR with the 🚨🚨 emoji so changelog authors can further highlight this if needed.

Pre-merge Checklist

Expand the relevant checklist and delete the others.

New Connector

Community member or Airbyter

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
If new credentials are required for use in CI, add them to GSM. Instructions.
/test connector=connectors/<name> command is passing
New Connector version released on Dockerhub by running the /publish command described here
After the connector is published, connector added to connector index as described here
Seed specs have been re-generated by building the platform and committing the changes to the seed spec files, as described here

Updating a connector

Community member or Airbyter

Grant edit access to maintainers (instructions)
Secrets in the connector's spec are annotated with airbyte_secret
Unit & integration tests added and passing. Community members, please provide proof of success locally e.g: screenshot or copy-paste unit, integration, and acceptance test output. To run acceptance tests for a Python connector, follow instructions in the README. For java connectors run ./gradlew :airbyte-integrations:connectors:<name>:integrationTest.
Code reviews completed
Documentation updated
- Connector's README.md
- Connector's bootstrap.md. See description and examples
- Changelog updated in docs/integrations/<source or destination>/<name>.md including changelog. See changelog example
PR name follows PR naming conventions

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
If new credentials are required for use in CI, add them to GSM. Instructions.
/test connector=connectors/<name> command is passing
New Connector version released on Dockerhub and connector version bumped by running the /publish command described here

Connector Generator

Issue acceptance criteria met
PR name follows PR naming conventions
If adding a new generator, add it to the list of scaffold modules being tested
The generator test modules (all connectors with -scaffold in their name) have been updated with the latest scaffold by running ./gradlew :airbyte-integrations:connector-templates:generator:testScaffoldTemplates then checking in your changes
Documentation which references the generator is updated as needed

Tests

Unit

Put your unit tests output here.

Integration

Put your integration tests output here.

Acceptance

Put your acceptance tests output here.

girarda · 2022-10-27T21:17:38Z

airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py

+                yield message
+                if message.type == MessageType.RECORD:
+                    record = message.record
+                    stream_state = stream_instance.get_updated_state(stream_state, record.data)


same logic moved inside the if block because we only want to update the state and record counter on records

girarda · 2022-10-27T21:18:28Z

airbyte-cdk/python/airbyte_cdk/sources/streams/core.py

@@ -99,6 +116,7 @@ def read_records(
        This method should be overridden by subclasses to read records based on the inputs
        """

+    @lru_cache(maxsize=None)


move caching from https://github.com/airbytehq/airbyte/pull/18572/files#diff-51c861828e0a614ef6ee21390f3ba17d7138549f0575051fdb24868ee05eabbbL301

girarda · 2022-10-27T21:20:02Z

airbyte-cdk/python/unit_tests/sources/test_source.py

@@ -425,8 +425,8 @@ def test_source_config_no_transform(abstract_source, catalog):
    records = [r for r in abstract_source.read(logger=logger_mock, config={}, catalog=catalog, state={})]
    assert len(records) == 2 * 5
    assert [r.record.data for r in records] == [{"value": 23}] * 2 * 5
-    assert http_stream.get_json_schema.call_count == 1
-    assert non_http_stream.get_json_schema.call_count == 1
+    assert http_stream.get_json_schema.call_count == 5


this assert verified that we used the cached transformer and schema instead of calling get_json_schema.
We now call get_json_schema more often, but the value is cached so the behavior is effectively the same

makes sense! but side question, could we add a test into core.py that calls get_json_schema() and asserts that some part of:

ResourceSchemaLoader(package_name_from_class(self.__class__)).get_schema(self.name)

only gets invoked once. like given that it's cached, we'd only expect self.__class__ or the underlying get_schema() to only get called once

done 0cbd28e

…18689)" This reverts commit bf06dec.

This reverts commit 5dac7c2.

girarda · 2022-11-05T01:10:19Z

airbyte-cdk/python/airbyte_cdk/sources/streams/core.py

+# AirbyteRecordMessage: An AirbyteRecordMessage
+# AirbyteLogMessage: A log message
+# AirbyteTraceMessage: A trace message
+StreamData = Union[Mapping[str, Any], AirbyteRecordMessage, AirbyteLogMessage, AirbyteTraceMessage]


I'm not attached to this typedef, but the full union definition is quite cumbersome

girarda · 2022-11-05T01:11:07Z

airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py

-        stream_instance = self._stream_to_instance_map[stream_name]
-        return stream_instance.transformer, stream_instance.get_json_schema()
-
-    def _as_airbyte_record(self, stream_name: str, data: Mapping[str, Any]):


extracted to a function so it can be reused by the SimpleRetriever

sherifnada

🚢

sherifnada · 2022-11-05T01:14:53Z

airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py

-from airbyte_cdk.sources.utils.transform import TypeTransformer
+from airbyte_cdk.sources.utils.record_helper import \
+    stream_data_to_airbyte_message
+from airbyte_cdk.sources.utils.schema_helpers import InternalConfig, \


did a formatter change? i don't recall seeing this before

nope. Ran gradlew format. I'll look into what's wrong with my setup

This reverts commit 2f91b80.

This reverts commit 62d95eb.

This reverts commit 27150ba.

brianjlai

lgtm!

…nent * master: 🪟 🎉 Enable frontend validation for <1hr syncs (cloud) #19028 Stream returns AirbyteMessages (#18572) 🎉 New Source - Recruitee [low-code SDK] (#18671) 🎉 New source: Breezometer [low-code cdk] (#18650) Check disabled connections after protocol update (#18990) Simple default replication worker refactor (#19002) 🎉 New Source: Visma e-conomic (#18595) 🎉 New Source: Fastbill (#18593) Bmoric/extract state api (#18980) 🎉 New Source: Zapier Supported Storage (#18442) 🎉 New source: Klarna (#18385) `AirbyteEstimateTraceMessage` (#18875) Extract source definition api (#18977) [low-code cdk] Allow for spec file to be defined in the yaml manifest instead of an external file (#18411) 🐛 Source HubSpot: fix property scopes (#18624) Ensure that only 6-character hex values are passed to monaco editor (#18943)

method yielding airbytemessage

5f88aa6

github-actions bot added the CDK Connector Development Kit label Oct 27, 2022

girarda added 2 commits October 27, 2022 13:33

move to Stream

0de8395

update abstract source

e5acdcf

girarda changed the title ~~method yielding airbytemessage~~ Stream returns AirbyteMessages Oct 27, 2022

girarda commented Oct 27, 2022

View reviewed changes

girarda added 22 commits October 27, 2022 14:20

reset

8a63af7

missing file

d0dcb52

add test docker image

adf07fa

script to run acceptance tests with local cdk

c68e01d

Update conftest to use a different image

258990d

extract to method

b0f80a7

dont use a different image tag

68528b2

Always install local cdk

b31f9bd

break the cdk

893beab

get path from current working directory

c3964c9

or

c69774a

ignore unit test

152f5df

debug log

8ea5e85

Revert "AMI change: ami-0f23be2f917510c26 -> ami-005924fb76f7477ce (#…

1684823

…18689)" This reverts commit bf06dec.

build from the top

b08a9c4

Update source-acceptance-test

4fd9ed0

fix

06b1c70

copy setup

3e4b135

some work on the gradle plugin

4702f4d

reset to master

ae7b7fb

delete unused file

3d86bce

delete unused file

9dc4486

girarda temporarily deployed to more-secrets November 5, 2022 00:33 Inactive

Revert "Revert "reset to master""

594ad06

This reverts commit 5dac7c2.

girarda commented Nov 5, 2022

View reviewed changes

girarda added 2 commits November 4, 2022 18:12

reset to master

2d91e9a

reset to master

cff30a1

github-actions bot removed the area/connectors Connector related issues label Nov 5, 2022

girarda requested review from sherifnada and brianjlai November 5, 2022 01:13

sherifnada approved these changes Nov 5, 2022

View reviewed changes

girarda mentioned this pull request Nov 5, 2022

SimpleRetriever yield request and response as log messages #18644

Merged

37 tasks

girarda added 7 commits November 4, 2022 18:39

test

2f91b80

Revert "test"

3fc697c

This reverts commit 2f91b80.

test

62d95eb

Revert "test"

003b860

This reverts commit 62d95eb.

test

27150ba

Revert "test"

26c3289

This reverts commit 27150ba.

format

b36842a

brianjlai approved these changes Nov 7, 2022

View reviewed changes

Merge branch 'master' into alex/read_with_logs

fe3b01a

girarda removed connectors/source/greenhouse connectors/source/sendgrid connectors/source/datadog connectors/source/alpha-vantage connectors/source/gridly labels Nov 8, 2022

girarda merged commit 1a608f8 into master Nov 8, 2022

girarda deleted the alex/read_with_logs branch November 8, 2022 05:23

github-actions bot mentioned this pull request Nov 15, 2022

Bump helm chart version reference to 0.40.43 #19453

Closed

octavia-squidington-iii mentioned this pull request Nov 18, 2022

Bump Airbyte version from 0.40.18 to 0.40.19 #19579

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Stream returns AirbyteMessages #18572

Stream returns AirbyteMessages #18572

girarda commented Oct 27, 2022 •

edited

Loading

girarda Oct 27, 2022 •

edited

Loading

girarda Oct 27, 2022

girarda Oct 27, 2022

brianjlai Nov 3, 2022

girarda Nov 5, 2022

girarda Nov 5, 2022

girarda Nov 5, 2022

sherifnada left a comment

sherifnada Nov 5, 2022

girarda Nov 5, 2022

brianjlai left a comment

Stream returns AirbyteMessages #18572

Stream returns AirbyteMessages #18572

Conversation

girarda commented Oct 27, 2022 • edited Loading

What

How

Testing

Recommended reading order

🚨 User Impact 🚨

Pre-merge Checklist

Community member or Airbyter

Airbyter

Community member or Airbyter

Airbyter

Tests

girarda Oct 27, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sherifnada left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brianjlai left a comment

Choose a reason for hiding this comment

girarda commented Oct 27, 2022 •

edited

Loading

girarda Oct 27, 2022 •

edited

Loading