CDK: Emit control message on config mutation #19428

alafanechere · 2022-11-15T14:33:19Z

What

Closes #17910
Quoting @davinchia in #3990:

Today we do not handle the case where a new refresh token is returned when a new access token is granted.

A new protocol message: AirbyteControlMessage.ConnectorConfig was introduced to enable the connector to share a configuration update with the platform.

This PR is an attempt to update the CDK to enable automatic emission of AirbyteControlMessage.ConnectorConfig on mutation of a config dict keys, and nested keys.

How

Add a config kwarg to OAuth2Authenticator.__init__. Observe this config by creating an ObservedDict assigned to self.connector_config
ObservedDict get created with a ConfigObserver object. ObservedDict call ConfigObserver.update on __set_item__ calls (this is the function called on key assignment: e.g config["refresh_token"] = "new_refresh_token".
ConfigObserver.update emits an AirbyteControlMessage.ConnectorConfig with the updated config.

N.B.

I chose the observer pattern because I thought this could be a silver-bullet-like approach to enable additional future use cases around AirbyteControlMessage.ConnectorConfig message.

The platform is not currently processing these new messages, so emitting will AirbyteControlMessage.ConnectorConfig will currently have no effect/

🚨 User Impact 🚨

To benefit from config observation, the connector developer can implement a custom authenticator inheriting from OAuth2Authenticator and create any token-refresh logic in this authenticator. The functions implementing this logic will have to mutate self.connector_config to emit a control message.

airbyte-cdk/python/airbyte_cdk/sources/utils/schema_helpers.py

airbyte-cdk/python/unit_tests/test_config_observation.py

sherifnada

@alafanechere could you show an example of how would this be used in the context of an HTTPStream and a single-use oauth authenticator?

alafanechere · 2022-11-16T15:51:40Z

@sherifnada I achieved a working example for BingAds in 733b89a (will revert afterward)

A READ outputs the following to stdout (I manually removed sensitive fields):

$ python main.py read --catalog integration_tests/configured_catalog.json --config secrets/config.json
{"type": "LOG", "log": {"level": "INFO", "message": "Starting syncing SourceBingAds"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Fetching access token ..."}}
{"type": "CONNECTOR_CONFIG", "emitted_at": 1668613492681.509, "connectorConfig": {"config": {"tenant_id": "common", "refresh_token": "<OBFUSCATED-FOR-DEMO>", "client_id": "<OBFUSCATED-FOR-DEMO>", "developer_token": "<OBFUSCATED-FOR-DEMO>", "reports_start_date": "2020-02-25", "hourly_reports": false, "daily_reports": false, "weekly_reports": true, "monthly_reports": true}}}
{"type": "LOG", "log": {"level": "INFO", "message": "Syncing stream: ad_groups "}}
... Records are coming in afterward

evantahler · 2022-11-16T16:16:27Z

{"type": "CONNECTOR_CONFIG", "emitted_at": 123, "connectorConfig": {...}}

I think there's one more envelope layer on that message, and it should be something like:

{"type": "CONTROL", "control": {"type": "CONNECTOR_CONFIG", "emitted_at": 123, "connectorConfig": {...}}}

alafanechere · 2022-11-16T16:17:24Z

{"type": "CONNECTOR_CONFIG", "emitted_at": 123, "connectorConfig": {...}}
I think there's one more envelope layer on that message, and it should be something like:
{"type": "CONTROL", "control": {"type": "CONNECTOR_CONFIG", "emitted_at": 123, "connectorConfig": {...}}}

I think I forgot to wrap this around an AirbyteMessage right?

evantahler · 2022-11-16T16:19:13Z

I think I forgot to wrap this around an AirbyteMessage right?

That's probably it!

alafanechere · 2022-11-23T16:06:37Z

@sherifnada I tried to go in the direction you suggested by implementing a specific SingleUseRefreshTokenOauth2Authenticator authenticator with a behavior that can match most existing OAuth scenarios. Let me know if it's what you had in mind 😄

alafanechere · 2022-11-23T16:07:14Z

I tried to use source-harvest as an example: c5a0b6d

sherifnada

nice! looking really good. Some questions

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

sherifnada · 2022-11-23T23:00:03Z

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

+            ValueError: Raised if the defined getters are not returning a value.
+        """
+        for field_name, getter in [
+            ("client_id", self.get_client_id),


why do we care about retaining and validating client_id and client_secret in this class instead of just passing up the concern to the super?

As client_id, client_secret, and refresh_token are mandatory arguments of the parent class, and the current constructor signature of the new class only takes a connector_config dict, I thought it would be safer to fail early if these fields' values could not be parsed from the config.

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

sherifnada · 2022-11-23T23:02:24Z

airbyte-integrations/connectors/source-harvest/source_harvest/source.py

                token_refresh_endpoint="https://id.getharvest.com/api/v2/oauth2/token",
-                client_id=credentials.get("client_id"),


why were these removed?

The client_id parameter does not exist in my SingleUseRefreshTokenOauth2Authenticator implementation. The client_id is retrieved from the configuration with dpath

sherifnada · 2022-11-23T23:04:01Z

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

+        try:
+            response_json = self._get_refresh_access_token_response()
+            return (
+                response_json[self.get_access_token_name()],


would recommend using dpath

Agreed but for implementation consistency with parent class and abstract class, I'd prefer to do this in a separate PR in which we can replace the access_token_name, expires_in_name, and refresh_token_name by dpaths that can be used when parsing these responses. Wdyt?

makes sense!

alafanechere · 2022-11-25T11:27:00Z

@sherifnada I used dpath to declare path to the config field that we use to build refresh requests and set a new refresh token values in the config. I think this is far better now for DX 😄
I think dpath could also be used to define with more flexibility the location of the fields we want to parse in the refresh token response, but I'd prefer to keep this refacto for a different PR.

sherifnada

Good stuff! requesting a couple small changes but no extra review needed from my side. Thank you for making this awesome change.

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

alafanechere · 2022-11-29T13:26:01Z

/publish-cdk dry-run=true

🕑 https://github.com/airbytehq/airbyte/actions/runs/3574511972
✅ https://github.com/airbytehq/airbyte/actions/runs/3574511972

alafanechere · 2022-11-29T13:37:43Z

/publish-cdk dry-run=false

🕑 https://github.com/airbytehq/airbyte/actions/runs/3574606697
✅ https://github.com/airbytehq/airbyte/actions/runs/3574606697

alafanechere added 2 commits November 15, 2022 00:53

wip

f3d4b5a

implementation

24d24dd

octavia-squidington-iv added the CDK Connector Development Kit label Nov 15, 2022

alafanechere added 3 commits November 15, 2022 16:00

format

355b90b

bump version

61e9d4b

Merge branch 'master' into augustin/cdk/emit-updated-configs

1101ccc

alafanechere marked this pull request as ready for review November 15, 2022 15:04

alafanechere requested a review from a team as a code owner November 15, 2022 15:04

alafanechere changed the title ~~CDK: Emit control message when config is mutated~~ CDK: Emit control message on config mutation Nov 15, 2022

alafanechere requested review from evantahler and a team November 15, 2022 15:05

erohmensing reviewed Nov 15, 2022

View reviewed changes

airbyte-cdk/python/airbyte_cdk/sources/utils/schema_helpers.py Outdated Show resolved Hide resolved

evantahler reviewed Nov 15, 2022

View reviewed changes

airbyte-cdk/python/unit_tests/test_config_observation.py Outdated Show resolved Hide resolved

sherifnada reviewed Nov 16, 2022

View reviewed changes

alafanechere added 2 commits November 16, 2022 15:27

always update by default, even if same value

22f75c7

rename split_config to filter_internal_keywords

d562117

octavia-squidington-iv added area/connectors Connector related issues connectors/source/hubspot connectors/source/mixpanel connectors/source/facebook-marketing connectors/source/amazon-ads connectors/source/azure-table connectors/source/google-search-console labels Nov 16, 2022

bing ads example

733b89a

octavia-squidington-iv added the connectors/source/bing-ads label Nov 16, 2022

wrap around AirbyteMessage

a59a1e6

alafanechere added 4 commits November 23, 2022 16:06

implement SingleUseRefreshTokenOauth2Authenticator

63d129c

test SingleUseRefreshTokenOauth2Authenticator

62ef553

call copy in ObservedDict

bbe802a

add docstring

8fd52f9

alafanechere force-pushed the augustin/cdk/emit-updated-configs branch from f5c3d0c to 8fd52f9 Compare November 23, 2022 15:45

source harvest example

c5a0b6d

octavia-squidington-iv added the connectors/source/harvest label Nov 23, 2022

alafanechere requested a review from sherifnada November 23, 2022 16:06

sherifnada reviewed Nov 23, 2022

View reviewed changes

alafanechere added 2 commits November 25, 2022 12:04

use dpath

803b70a

better doc string

37f474f

alafanechere requested a review from sherifnada November 25, 2022 11:25

update changelog

08f64e4

sherifnada approved these changes Nov 29, 2022

View reviewed changes

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py Outdated Show resolved Hide resolved

airbyte-cdk/python/airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py Outdated Show resolved Hide resolved

alafanechere added 3 commits November 29, 2022 11:53

use sequence instead of string path for dpath declaration

518a36f

Merge branch 'master' into augustin/cdk/emit-updated-configs

64bbc48

revert connector changes

4da6581

octavia-squidington-iv removed the area/connectors Connector related issues label Nov 29, 2022

format

380e175

alafanechere merged commit 3b15f9e into master Nov 29, 2022

alafanechere deleted the augustin/cdk/emit-updated-configs branch November 29, 2022 13:51

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

Bump Airbyte version from 0.40.22 to 0.40.23 #19941

Merged

alafanechere mentioned this pull request Dec 2, 2022

SAT: Capture configuration updates from connectors' control messages #19979

Merged

joelluijmes mentioned this pull request Dec 15, 2022

🎉 New Source: Exact Online #20480

Closed

11 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

CDK: Emit control message on config mutation #19428

CDK: Emit control message on config mutation #19428

alafanechere commented Nov 15, 2022 •

edited

Loading

sherifnada left a comment

alafanechere commented Nov 16, 2022

evantahler commented Nov 16, 2022 •

edited

Loading

alafanechere commented Nov 16, 2022

evantahler commented Nov 16, 2022 •

edited

Loading

alafanechere commented Nov 23, 2022

alafanechere commented Nov 23, 2022

sherifnada left a comment

sherifnada Nov 23, 2022

alafanechere Nov 25, 2022 •

edited

Loading

sherifnada Nov 23, 2022

alafanechere Nov 25, 2022

sherifnada Nov 23, 2022

alafanechere Nov 25, 2022

sherifnada Nov 29, 2022 •

edited

Loading

alafanechere commented Nov 25, 2022 •

edited

Loading

sherifnada left a comment

alafanechere commented Nov 29, 2022 •

edited by github-actions bot

Loading

alafanechere commented Nov 29, 2022 •

edited by github-actions bot

Loading

		token_refresh_endpoint="https://id.getharvest.com/api/v2/oauth2/token",
		client_id=credentials.get("client_id"),

CDK: Emit control message on config mutation #19428

CDK: Emit control message on config mutation #19428

Conversation

alafanechere commented Nov 15, 2022 • edited Loading

What

How

N.B.

Recommended reading order

🚨 User Impact 🚨

sherifnada left a comment

Choose a reason for hiding this comment

alafanechere commented Nov 16, 2022

evantahler commented Nov 16, 2022 • edited Loading

alafanechere commented Nov 16, 2022

evantahler commented Nov 16, 2022 • edited Loading

alafanechere commented Nov 23, 2022

alafanechere commented Nov 23, 2022

sherifnada left a comment

Choose a reason for hiding this comment

sherifnada Nov 23, 2022

Choose a reason for hiding this comment

alafanechere Nov 25, 2022 • edited Loading

Choose a reason for hiding this comment

sherifnada Nov 23, 2022

Choose a reason for hiding this comment

alafanechere Nov 25, 2022

Choose a reason for hiding this comment

sherifnada Nov 23, 2022

Choose a reason for hiding this comment

alafanechere Nov 25, 2022

Choose a reason for hiding this comment

sherifnada Nov 29, 2022 • edited Loading

Choose a reason for hiding this comment

alafanechere commented Nov 25, 2022 • edited Loading

sherifnada left a comment

Choose a reason for hiding this comment

alafanechere commented Nov 29, 2022 • edited by github-actions bot Loading

alafanechere commented Nov 29, 2022 • edited by github-actions bot Loading

alafanechere commented Nov 15, 2022 •

edited

Loading

evantahler commented Nov 16, 2022 •

edited

Loading

evantahler commented Nov 16, 2022 •

edited

Loading

alafanechere Nov 25, 2022 •

edited

Loading

sherifnada Nov 29, 2022 •

edited

Loading

alafanechere commented Nov 25, 2022 •

edited

Loading

alafanechere commented Nov 29, 2022 •

edited by github-actions bot

Loading

alafanechere commented Nov 29, 2022 •

edited by github-actions bot

Loading