[8/n][dagster-fivetran] Implement FivetranConnector and FivetranDestination #25889

maximearmstrong · 2024-11-12T22:56:41Z

Summary & Motivation

This PR implements FivetranConnector and FivetranDestination, and removes FivetranContentData and FivetranContentType.

This addresses the concerns raised here about the legacy code.

How I Tested These Changes

BK with same tests.

maximearmstrong · 2024-11-12T22:56:54Z

This stack of pull requests is managed by Graphite. Learn more about stacking.

maximearmstrong · 2024-11-12T23:08:31Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/asset_defs.py

@@ -127,7 +131,9 @@ def _build_fivetran_assets(
                connector_id=connection_metadata.connector_id,
                name=connection_metadata.name,
                connector_url=connection_metadata.connector_url,
-                schema_config=connection_metadata.schemas,
+                schema_config=FivetranSchemaConfig.from_schema_config_details(


[1] FivetranConnectorTableProps was updated to take a FivetranSchemaConfig object instead of a Mapping[str, Any] one. The goal here is to keep FivetranConnectorTableProps as close as possible to the new implementation.

Given that the two implementations are built across the same version of fivetran's API, it makes sense to me that we should be able to share implementation across these subclasses.

maximearmstrong · 2024-11-12T23:09:18Z

python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/utils.py

@@ -122,6 +122,9 @@ def get_complex_sample_connector_schema_config(
                                "allowed": False,
                                "reason_code": "SYSTEM_TABLE",
                            },
+                            "columns": {


[1] Columns are required in the schema config API docs. By updating the legacy test samples, the goal is that the legacy tests succeed with the updated FivetranConnectorTableProps.

Historically this wasn't the case (#23631) and if the user did not customize which columns to sync for a table, the Fivetran API would not return column schema info - might be good to verify that this is not the case now and/or still make sure we have a fallback in case we don't get column level info

That's good feedback - I reverted the test changes in a171504 and made columns optional

maximearmstrong · 2024-11-12T23:10:33Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/utils.py

@@ -23,6 +23,7 @@ def get_fivetran_connector_table_name(schema_name: str, table_name: str) -> str:
    return f"{schema_name}.{table_name}"


+# TODO: move to translator.py and make table_data a FivetranTable object


TODO added because this is out-of-scope for this PR, but should be addressed when the legacy code is removed.

Nit; let's not leave todos in the code. At least 50% of the time they end up staying there 😢 - any time there's a todo, I feel like it should probably be a linear ticket (or subticket)

Very fair - removed in a1cea87 and I created a linear ticket

graphite-app · 2024-11-12T23:25:37Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/translator.py

+        connector_details: Mapping[str, Any],
+        destination_details: Mapping[str, Any],
+        schema_config_details: Mapping[str, Any],
+    ) -> "FivetranDestination":


The return type annotation is incorrect - this method returns a FivetranConnector instance but is annotated to return FivetranDestination. The annotation should be updated to -> "FivetranConnector".

Spotted by Graphite Reviewer

Is this helpful? React 👍 or 👎 to let us know.

benpankow

will leave for @dpeng817 to approve but looks good to me - left one quick comment about column schema which historically hasn't been included in all API responses - this has caused us troubl ein the past, might not be the case anymore though

benpankow · 2024-11-13T00:10:10Z

python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/utils.py

@@ -122,6 +122,9 @@ def get_complex_sample_connector_schema_config(
                                "allowed": False,
                                "reason_code": "SYSTEM_TABLE",
                            },
+                            "columns": {


Historically this wasn't the case (#23631) and if the user did not customize which columns to sync for a table, the Fivetran API would not return column schema info - might be good to verify that this is not the case now and/or still make sure we have a fallback in case we don't get column level info

dpeng817

a few comments + the test behavior change is a bit scary, i think we should aim to, within this PR, not have to update tests at all

dpeng817 · 2024-11-13T16:13:02Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/translator.py

-        for connector_id, connector_data in self.connectors_by_id.items():
-            connector_details = connector_data.properties
-            connector_name = connector_details["schema"]
-            connector_url = get_fivetran_connector_url(connector_details)
-
-            schema_config = connector_details["schema_config"]
-
-            destination_details = self.destinations_by_id[
-                connector_details["destination_id"]
-            ].properties
-            database = destination_details.get("config", {}).get("database")
-            service = destination_details.get("service")
-
-            schemas_data = cast(Dict[str, Any], schema_config["schemas"])
-            for schema in schemas_data.values():
-                if schema["enabled"]:
-                    schema_name = schema["name_in_destination"]
-                    schema_tables: Dict[str, Dict[str, Any]] = cast(
-                        Dict[str, Dict[str, Any]], schema["tables"]
-                    )
-                    for table in schema_tables.values():
-                        if table["enabled"]:
-                            table_name = table["name_in_destination"]


dpeng817 · 2024-11-13T16:14:50Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/utils.py

@@ -23,6 +23,7 @@ def get_fivetran_connector_table_name(schema_name: str, table_name: str) -> str:
    return f"{schema_name}.{table_name}"


+# TODO: move to translator.py and make table_data a FivetranTable object


Nit; let's not leave todos in the code. At least 50% of the time they end up staying there 😢 - any time there's a todo, I feel like it should probably be a linear ticket (or subticket)

dpeng817 · 2024-11-13T16:17:22Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/asset_defs.py

@@ -127,7 +131,9 @@ def _build_fivetran_assets(
                connector_id=connection_metadata.connector_id,
                name=connection_metadata.name,
                connector_url=connection_metadata.connector_url,
-                schema_config=connection_metadata.schemas,
+                schema_config=FivetranSchemaConfig.from_schema_config_details(


Given that the two implementations are built across the same version of fivetran's API, it makes sense to me that we should be able to share implementation across these subclasses.

dpeng817 · 2024-11-13T16:23:19Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/resources.py

            destination_details = client.get_destination_details(destination_id=group_id)
-            destinations.append(
-                FivetranContentData(
-                    content_type=FivetranContentType.DESTINATION, properties=destination_details
-                )
+            destination_id = destination_details["id"]
+
+            destinations_by_id[destination_id] = FivetranDestination.from_destination_details(
+                destination_details=destination_details


hm feels like all this string level munging should be avoidable now right? You should be able to directly construct a FivetranDestination, FivetranConnector, etc right and avoid all the bare indexing as a result?

Updated in 6e42757

maximearmstrong · 2024-11-13T16:35:52Z

a few comments + the test behavior change is a bit scary, i think we should aim to, within this PR, not have to update tests at all

@dpeng817 That's fair - I can revert the changes for the tests and support optional columns. Fivetran API reference states that columns are required in the schema config, but we can that when the the legacy code is deprecated.

re-requesting

dpeng817 · 2024-11-13T19:45:00Z

@maximearmstrong yea I think the change is probably reasonable, but generally good to consider that type of thing in isolation I think.

dpeng817

Looks good, a few design proposals for your consideration. Will leave it up to you whether you want to go forward with them.

dpeng817 · 2024-11-13T19:46:36Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/resources.py

                )

-        return FivetranWorkspaceData.from_content_data(connectors + destinations)
+                if not connector.has_bad_status:


I know I sent this property name privately, but definitely open to other names 😆

I updated the property to is_connected in 5d9e68b

dpeng817 · 2024-11-13T19:49:25Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/translator.py

+        schema_config_details: Mapping[str, Any],
+    ) -> "FivetranConnector":
+        return cls(
+            id=connector_details["id"],


is the connector_details dictionary large? I think my preference is for the record to have just connector_details and schema_config as an attribute, and then destination_id, has_bad_status etc could all just be properties built off of it. But if the dictionary is too large, then I understand.

Given that this is only being persisted in the cache, I view it as a reversible decision either way, and will leave that up to you.

The dictionary is quite large. That said, I think you are right and values like has_bad_status should be properties.

Now working on the next PR to implement sync and poll methods, we often use the connector details to test conditions. In hindsight, I believe FivetranConnector should be built only using the connector details retrieved from the API - the schema configs should be decoupled and stored in the FivetranWorkspaceData object. I will update the PR to reflect these changes.

Code updated in 6685d1b

dpeng817 · 2024-11-13T19:50:38Z

python_modules/libraries/dagster-fivetran/dagster_fivetran/translator.py

+    service: str
+    schema_config: "FivetranSchemaConfig"
+    destination_id: str
+    has_bad_status: bool


If we do end up making this an attribute instead of a property, I think I would have a preference for storing the setup_state directly as an attribute, and then you can have is_invalid_state or something as a property on the enum.

…nation

maximearmstrong mentioned this pull request Nov 12, 2024

[7/n][dagster-fivetran] Implement load_fivetran_asset_specs #25808

Merged

maximearmstrong commented Nov 12, 2024

View reviewed changes

graphite-app bot reviewed Nov 12, 2024

View reviewed changes

maximearmstrong self-assigned this Nov 12, 2024

maximearmstrong marked this pull request as ready for review November 12, 2024 23:41

maximearmstrong requested review from dpeng817, benpankow and alangenfeld November 12, 2024 23:41

benpankow reviewed Nov 13, 2024

View reviewed changes

maximearmstrong force-pushed the maxime/rework-fivetran-7 branch from 0923217 to 14f49b7 Compare November 13, 2024 12:50

maximearmstrong force-pushed the maxime/rework-fivetran-8 branch from 4418da5 to 0997b15 Compare November 13, 2024 12:50

Base automatically changed from maxime/rework-fivetran-7 to master November 13, 2024 13:08

maximearmstrong force-pushed the maxime/rework-fivetran-8 branch from 0997b15 to 47a8a9d Compare November 13, 2024 13:09

dpeng817 previously requested changes Nov 13, 2024

View reviewed changes

maximearmstrong force-pushed the maxime/rework-fivetran-8 branch from 47a8a9d to 6e42757 Compare November 13, 2024 19:11

maximearmstrong requested review from dpeng817 and benpankow November 13, 2024 19:13

dpeng817 approved these changes Nov 13, 2024

View reviewed changes

maximearmstrong added 14 commits November 14, 2024 21:12

Add get_fivetran_connector_table_name to utils

d3153ad

Move api_key and api_secret mocks to conftest

8f009e2

Lint

d35d3d9

[8/n][dagster-fivetran] Implement FivetranConnector and FivetranDesti…

cea52d9

…nation

Fix merge

f914e99

Add builder fn

6cc879f

Update FivetranDestination

3d4f1c9

Update FivetranConnector

df94068

Revert legacy tests, make FivetranTable.columns optional

ee9350b

Remove todo

d63c02b

move connector status logic to FivetranConnector

c436c46

Decouple schema config from connector, implement connector properties

854e3a0

Update connector setup state property

ba28c05

Store setup state as string

283f3d8

maximearmstrong force-pushed the maxime/rework-fivetran-8 branch from a8daee3 to 283f3d8 Compare November 15, 2024 02:12

maximearmstrong mentioned this pull request Nov 15, 2024

[10/n][dagster-fivetran] Implement fivetran_assets and build_fivetran_assets_definitions #25944

Merged

maximearmstrong merged commit 31f7566 into master Nov 15, 2024
1 check failed

maximearmstrong deleted the maxime/rework-fivetran-8 branch November 15, 2024 14:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[8/n][dagster-fivetran] Implement FivetranConnector and FivetranDestination #25889

[8/n][dagster-fivetran] Implement FivetranConnector and FivetranDestination #25889

maximearmstrong commented Nov 12, 2024 •

edited

Loading

maximearmstrong commented Nov 12, 2024 •

edited

Loading

maximearmstrong Nov 12, 2024

dpeng817 Nov 13, 2024

maximearmstrong Nov 12, 2024

benpankow Nov 13, 2024

maximearmstrong Nov 13, 2024

maximearmstrong Nov 12, 2024 •

edited

Loading

dpeng817 Nov 13, 2024

maximearmstrong Nov 13, 2024

graphite-app bot Nov 12, 2024

benpankow left a comment

benpankow Nov 13, 2024

dpeng817 left a comment

dpeng817 Nov 13, 2024

dpeng817 Nov 13, 2024

dpeng817 Nov 13, 2024

dpeng817 Nov 13, 2024

maximearmstrong Nov 13, 2024

maximearmstrong commented Nov 13, 2024

dpeng817 commented Nov 13, 2024

dpeng817 left a comment

dpeng817 Nov 13, 2024

maximearmstrong Nov 13, 2024

dpeng817 Nov 13, 2024

maximearmstrong Nov 13, 2024

maximearmstrong Nov 13, 2024

dpeng817 Nov 13, 2024

		@@ -23,6 +23,7 @@ def get_fivetran_connector_table_name(schema_name: str, table_name: str) -> str:
		return f"{schema_name}.{table_name}"


		# TODO: move to translator.py and make table_data a FivetranTable object

[8/n][dagster-fivetran] Implement FivetranConnector and FivetranDestination #25889

[8/n][dagster-fivetran] Implement FivetranConnector and FivetranDestination #25889

Conversation

maximearmstrong commented Nov 12, 2024 • edited Loading

Summary & Motivation

How I Tested These Changes

maximearmstrong commented Nov 12, 2024 • 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

maximearmstrong Nov 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graphite-app bot Nov 12, 2024

Choose a reason for hiding this comment

benpankow left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dpeng817 left a 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

maximearmstrong commented Nov 13, 2024

dpeng817 commented Nov 13, 2024

dpeng817 left a 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

Choose a reason for hiding this comment

maximearmstrong commented Nov 12, 2024 •

edited

Loading

maximearmstrong commented Nov 12, 2024 •

edited

Loading

maximearmstrong Nov 12, 2024 •

edited

Loading