add resync and poll (#32734)

cnolanminich · web-flow · commit 10ebdb5e5642 · 2025-11-14T17:34:25.000-05:00
diff --git a/docs/docs/integrations/libraries/fivetran/fivetran-pythonic.md b/docs/docs/integrations/libraries/fivetran/fivetran-pythonic.md
@@ -84,6 +84,29 @@ If you want to customize the sync of your connectors, you can use the <PyObject
   language="python"
 />
 
+### Perform historical resyncs of Fivetran assets
+
+In addition to incremental syncs, you can perform full historical resyncs of your Fivetran connector tables using the <PyObject section="libraries" module="dagster_fivetran" object="FivetranWorkspace" displayText="resync_and_poll()" /> method. This is useful when you need to backfill historical data or reload data after schema changes.
+
+You can resync specific tables by providing `resync_parameters`, or resync all tables in a connector by omitting this parameter:
+
+<CodeExample path="docs_snippets/docs_snippets/integrations/fivetran/resync_fivetran_assets.py" language="python" />
+
+To resync all tables in a connector, simply call `resync_and_poll()` without the `resync_parameters` argument:
+
+<CodeExample
+  startAfter="start_resync_all"
+  endBefore="end_resync_all"
+  path="docs_snippets/docs_snippets/integrations/fivetran/resync_fivetran_assets.py"
+  language="python"
+/>
+
+:::note
+
+Historical resyncs can be time-consuming and resource-intensive operations. Be mindful of the cost implications when resyncing large datasets.
+
+:::
+
 ### Customize asset definition metadata for Fivetran assets
 
 By default, Dagster will generate asset specs for each Fivetran asset and populate default metadata. You can further customize asset properties by passing an instance of the custom <PyObject section="libraries" module="dagster_fivetran" object="DagsterFivetranTranslator" /> to the <PyObject section="libraries" module="dagster_fivetran" object="load_fivetran_asset_specs" /> function.
diff --git a/examples/docs_snippets/docs_snippets/integrations/fivetran/resync_fivetran_assets.py b/examples/docs_snippets/docs_snippets/integrations/fivetran/resync_fivetran_assets.py
@@ -0,0 +1,58 @@
+from dagster_fivetran import FivetranSyncConfig, FivetranWorkspace, fivetran_assets
+
+import dagster as dg
+
+fivetran_workspace = FivetranWorkspace(
+    account_id=dg.EnvVar("FIVETRAN_ACCOUNT_ID"),
+    api_key=dg.EnvVar("FIVETRAN_API_KEY"),
+    api_secret=dg.EnvVar("FIVETRAN_API_SECRET"),
+)
+
+
+@fivetran_assets(
+    connector_id="fivetran_connector_id",  # Replace with your connector ID
+    name="fivetran_connector_name",  # Replace with your connection name
+    group_name="fivetran_connector_name",
+    workspace=fivetran_workspace,
+)
+def fivetran_connector_assets(
+    context: dg.AssetExecutionContext,
+    fivetran: FivetranWorkspace,
+    config: FivetranSyncConfig,
+):
+    """Syncs Fivetran connector with optional resync capability.
+
+    Configure at runtime:
+    - For normal sync: Pass config with resync=False (default)
+    - For historical resync of specific tables: Pass config with resync=True and resync_parameters
+    - For full historical resync: Pass config with resync=True and no resync_parameters
+    """
+    yield from fivetran.sync_and_poll(context=context, config=config)
+
+
+# start_resync_all
+@fivetran_assets(
+    connector_id="fivetran_connector_id",
+    name="fivetran_connector_name_full_resync",
+    group_name="fivetran_connector_name",
+    workspace=fivetran_workspace,
+)
+def fivetran_connector_full_resync_assets(
+    context: dg.AssetExecutionContext,
+    fivetran: FivetranWorkspace,
+    config: FivetranSyncConfig,
+):
+    """Performs a full historical resync of all tables in the connector.
+
+    Configure at runtime with resync=True.
+    """
+    yield from fivetran.sync_and_poll(context=context, config=config)
+
+
+# end_resync_all
+
+
+defs = dg.Definitions(
+    assets=[fivetran_connector_assets, fivetran_connector_full_resync_assets],
+    resources={"fivetran": fivetran_workspace},
+)
diff --git a/examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py b/examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py
@@ -34,6 +34,7 @@
     f"{snippets_folder}/fivetran/create_fivetran_all_assets_job.py",
     f"{snippets_folder}/fivetran/create_fivetran_selection_job.py",
     f"{snippets_folder}/fivetran/schedule_fivetran_jobs.py",
+    f"{snippets_folder}/fivetran/resync_fivetran_assets.py",
     f"{snippets_folder}/airbyte_cloud/customize_airbyte_cloud_asset_defs.py",
     f"{snippets_folder}/airbyte_cloud/customize_airbyte_cloud_translator_asset_spec.py",
     f"{snippets_folder}/airbyte_cloud/define_downstream_dependencies.py",
diff --git a/python_modules/libraries/dagster-fivetran/dagster_fivetran/__init__.py b/python_modules/libraries/dagster-fivetran/dagster_fivetran/__init__.py
@@ -15,6 +15,7 @@
 )
 from dagster_fivetran.resources import (
     FivetranResource as FivetranResource,
+    FivetranSyncConfig as FivetranSyncConfig,
     FivetranWorkspace as FivetranWorkspace,
     fivetran_resource as fivetran_resource,
     load_fivetran_asset_specs as load_fivetran_asset_specs,
diff --git a/python_modules/libraries/dagster-fivetran/dagster_fivetran/resources.py b/python_modules/libraries/dagster-fivetran/dagster_fivetran/resources.py
@@ -13,6 +13,7 @@
 from dagster import (
     AssetExecutionContext,
     AssetMaterialization,
+    Config,
     Definitions,
     Failure,
     InitResourceContext,
@@ -928,6 +929,29 @@ def _sync_and_poll(
         return FivetranOutput(connector_details=final_details, schema_config=schema_config_details)
 
 
+class FivetranSyncConfig(Config):
+    """Configuration for controlling Fivetran sync behavior.
+
+    Attributes:
+        resync: If True, performs a historical resync. If False, performs a normal sync.
+        resync_parameters: Optional parameters to control which tables to resync.
+            If not provided with resync=True, all tables will be resynced.
+            Example: {"schema_name": ["table1", "table2"], "another_schema": ["table3"]}
+    """
+
+    resync: bool = Field(
+        default=False,
+        description="Whether to perform a historical resync instead of a normal sync",
+    )
+    resync_parameters: Optional[dict[str, Any]] = Field(
+        default=None,
+        description=(
+            "Optional parameters to control which tables to resync. "
+            "If not provided with resync=True, all tables will be resynced."
+        ),
+    )
+
+
 class FivetranWorkspace(ConfigurableResource):
     """This class represents a Fivetran workspace and provides utilities
     to interact with Fivetran APIs.
@@ -1185,22 +1209,84 @@ def _generate_materialization(
 
     @public
     def sync_and_poll(
-        self, context: AssetExecutionContext
+        self,
+        context: AssetExecutionContext,
+        config: Optional[FivetranSyncConfig] = None,
     ) -> FivetranEventIterator[Union[AssetMaterialization, MaterializeResult]]:
         """Executes a sync and poll process to materialize Fivetran assets.
             This method can only be used in the context of an asset execution.
 
         Args:
             context (AssetExecutionContext): The execution context
                 from within `@fivetran_assets`.
+            config (Optional[FivetranSyncConfig]): Optional configuration to control sync behavior.
+                If config.resync is True, performs a historical resync instead of a normal sync.
+                If config.resync_parameters is provided, only the specified tables will be resynced.
 
         Returns:
             Iterator[Union[AssetMaterialization, MaterializeResult]]: An iterator of MaterializeResult
                 or AssetMaterialization.
+
+        Examples:
+            Normal sync (without config):
+
+            .. code-block:: python
+
+                from dagster import AssetExecutionContext
+                from dagster_fivetran import FivetranWorkspace, fivetran_assets
+
+                @fivetran_assets(connector_id="my_connector", workspace=fivetran_workspace)
+                def my_fivetran_assets(context: AssetExecutionContext, fivetran: FivetranWorkspace):
+                    yield from fivetran.sync_and_poll(context=context)
+
+            Historical resync of specific tables (config passed at runtime):
+
+            .. code-block:: python
+
+                from dagster import AssetExecutionContext
+                from dagster_fivetran import FivetranWorkspace, FivetranSyncConfig, fivetran_assets
+
+                @fivetran_assets(connector_id="my_connector", workspace=fivetran_workspace)
+                def my_fivetran_assets(
+                    context: AssetExecutionContext,
+                    fivetran: FivetranWorkspace,
+                    config: FivetranSyncConfig,
+                ):
+                    # When materializing, pass config with:
+                    # resync=True
+                    # resync_parameters={"schema_name": ["table1", "table2"]}
+                    yield from fivetran.sync_and_poll(context=context, config=config)
+
+            Full historical resync (config passed at runtime):
+
+            .. code-block:: python
+
+                from dagster import AssetExecutionContext
+                from dagster_fivetran import FivetranWorkspace, FivetranSyncConfig, fivetran_assets
+
+                @fivetran_assets(connector_id="my_connector", workspace=fivetran_workspace)
+                def my_fivetran_assets(
+                    context: AssetExecutionContext,
+                    fivetran: FivetranWorkspace,
+                    config: FivetranSyncConfig,
+                ):
+                    # When materializing, pass config with resync=True to resync all tables
+                    yield from fivetran.sync_and_poll(context=context, config=config)
         """
-        return FivetranEventIterator(
-            events=self._sync_and_poll(context=context), fivetran_workspace=self, context=context
-        )
+        if config and config.resync:
+            return FivetranEventIterator(
+                events=self._resync_and_poll(
+                    context=context, resync_parameters=config.resync_parameters
+                ),
+                fivetran_workspace=self,
+                context=context,
+            )
+        else:
+            return FivetranEventIterator(
+                events=self._sync_and_poll(context=context),
+                fivetran_workspace=self,
+                context=context,
+            )
 
     def _sync_and_poll(self, context: AssetExecutionContext):
         assets_def = context.assets_def
@@ -1245,6 +1331,54 @@ def _sync_and_poll(self, context: AssetExecutionContext):
         if unmaterialized_asset_keys:
             context.log.warning(f"Assets were not materialized: {unmaterialized_asset_keys}")
 
+    def _resync_and_poll(
+        self,
+        context: AssetExecutionContext,
+        resync_parameters: Optional[Mapping[str, Sequence[str]]] = None,
+    ):
+        assets_def = context.assets_def
+        dagster_fivetran_translator = get_translator_from_fivetran_assets(assets_def)
+        connector_id = next(
+            check.not_none(FivetranMetadataSet.extract(spec.metadata).connector_id)
+            for spec in assets_def.specs
+        )
+
+        client = self.get_client()
+        fivetran_output = client.resync_and_poll(
+            connector_id=connector_id,
+            resync_parameters=resync_parameters,
+        )
+
+        # The FivetranOutput is None if the connector hasn't been synced
+        if not fivetran_output:
+            context.log.warning(
+                f"The connector with ID {connector_id} is currently paused and so it has not been resynced. "
+                f"Make sure that your connector is enabled before resyncing it with Dagster."
+            )
+            return
+
+        materialized_asset_keys = set()
+        for materialization in self._generate_materialization(
+            fivetran_output=fivetran_output, dagster_fivetran_translator=dagster_fivetran_translator
+        ):
+            # Scan through all tables actually created, if it was expected then emit a MaterializeResult.
+            # Otherwise, emit a runtime AssetMaterialization.
+            if materialization.asset_key in context.selected_asset_keys:
+                yield MaterializeResult(
+                    asset_key=materialization.asset_key, metadata=materialization.metadata
+                )
+                materialized_asset_keys.add(materialization.asset_key)
+            else:
+                context.log.warning(
+                    f"An unexpected asset was materialized: {materialization.asset_key}. "
+                    f"Yielding a materialization event."
+                )
+                yield materialization
+
+        unmaterialized_asset_keys = context.selected_asset_keys - materialized_asset_keys
+        if unmaterialized_asset_keys:
+            context.log.warning(f"Assets were not materialized: {unmaterialized_asset_keys}")
+
 
 def load_fivetran_asset_specs(
     workspace: FivetranWorkspace,
diff --git a/python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/conftest.py b/python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/conftest.py
@@ -775,3 +775,41 @@ def sync_and_poll_fixture():
             paused_connector_output,
         ]
         yield mocked_function
+
+
+@pytest.fixture(name="resync_and_poll")
+def resync_and_poll_fixture():
+    with (
+        patch("dagster_fivetran.resources.FivetranClient.resync_and_poll") as mocked_function,
+        patch(
+            "dagster_fivetran.resources.FivetranResource.resync_and_poll"
+        ) as mocked_resync_and_poll_legacy_resource,
+    ):
+        # Fivetran output where all resync'd tables match the workspace data that was used to create the assets def
+        expected_fivetran_output = FivetranOutput(
+            connector_details=get_sample_connection_details(
+                succeeded_at=TEST_MAX_TIME_STR, failed_at=TEST_PREVIOUS_MAX_TIME_STR
+            )["data"],
+            schema_config=SAMPLE_SCHEMA_CONFIG_FOR_CONNECTOR["data"],
+        )
+        # Fivetran output where a table is missing and an unexpected table is resync'd,
+        # compared to the workspace data that was used to create the assets def
+        unexpected_fivetran_output = FivetranOutput(
+            connector_details=get_sample_connection_details(
+                succeeded_at=TEST_MAX_TIME_STR, failed_at=TEST_PREVIOUS_MAX_TIME_STR
+            )["data"],
+            schema_config=ALTERED_SAMPLE_SCHEMA_CONFIG_FOR_CONNECTOR["data"],
+        )
+        # resync_and_poll returns None if the connector is paused
+        paused_connector_output = None
+        mocked_function.side_effect = [
+            expected_fivetran_output,
+            unexpected_fivetran_output,
+            paused_connector_output,
+        ]
+        mocked_resync_and_poll_legacy_resource.side_effect = [
+            expected_fivetran_output,
+            unexpected_fivetran_output,
+            paused_connector_output,
+        ]
+        yield mocked_function
diff --git a/python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/test_resources.py b/python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/test_resources.py

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,7 @@`
`15`	`15`	`)`
`16`	`16`	`from dagster_fivetran.resources import (`
`17`	`17`	`FivetranResource as FivetranResource,`
	`18`	`+ FivetranSyncConfig as FivetranSyncConfig,`
`18`	`19`	`FivetranWorkspace as FivetranWorkspace,`
`19`	`20`	`fivetran_resource as fivetran_resource,`
`20`	`21`	`load_fivetran_asset_specs as load_fivetran_asset_specs,`