[11/n][dagster-fivetran] Implement materialization method in FivetranWorkspace

[maximearmstrong]

Summary & Motivation

This PR implements FivetranWorkspace.sync_and_poll, the materialization method for Fivetran assets. This method:

• calls FivetranClient.sync_and_poll
• takes the FivetranOutput returned by FivetranClient.sync_and_poll and generates the asset materializations
• yields MaterializeResult for each expected asset and AssetMaterialization for each unexpected asset
  • a connector table that was not in the connector at definitions loading time can be in the FivetranOutput. Eg. the table was added after definitions loading time and before sync.
• logs a warning for each unmaterialized table
  • a connector table can be created at definitions loading time, but can be missing in the FivetranOutput. Eg. the table was deleted after definitions loading time and before sync.

Can be leveraged like:

from dagster_fivetran import FivetranWorkspace, fivetran_assets

import dagster as dg

fivetran_workspace = FivetranWorkspace(
    account_id=dg.EnvVar("FIVETRAN_API_KEY"),
    api_key=dg.EnvVar("FIVETRAN_API_KEY"),
    api_secret=dg.EnvVar("FIVETRAN_API_SECRET"),
)


@fivetran_assets(
    connector_id="connector_id",
    name="connector_id",
    group_name="connector_id",
    workspace=fivetran_workspace,
)
def fivetran_connector_assets(context: dg.AssetExecutionContext, fivetran: FivetranWorkspace):
    yield from fivetran.sync_and_poll(context=context)


defs = dg.Definitions(
    assets=[fivetran_connector_assets],
    resources={"fivetran": fivetran_workspace},
)

How I Tested These Changes

Additional tests with BK

Tested with a live Fivetran instance:

Changelog

[dagster-fivetran]

[maximearmstrong]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• [11/n][dagster-fivetran] Implement materialization method in FivetranWorkspace #25961 : 2 dependent PRs (#26009 , #26181 ) 👈 (View in Graphite)
• [dagster-fivetran] Use Fivetran translator instance in load specs fn and state-backed defs #26133
• [10/n][dagster-fivetran] Implement fivetran_assets and build_fivetran_assets_definitions #25944
• [dagster-fivetran] Implement resync_and_poll method in FivetranClient #26062
• [dagster-fivetran] Implement sync_and_poll method in FivetranClient #26061
• [dagster-fivetran] Implement base poll method in FivetranClient #26060
• [dagster-fivetran] Implement base resync method in FivetranClient #26059
• [9/n][dagster-fivetran] Implement base sync methods in FivetranClient #25911
• [8/n][dagster-fivetran] Implement FivetranConnector and FivetranDestination #25889
• [7/n][dagster-fivetran] Implement load_fivetran_asset_specs #25808
• [6/n][dagster-fivetran] Implement FivetranWorkspaceDefsLoader #25807
• [5/n][dagster-fivetran] Implement FivetranWorkspaceData to FivetranConnectorTableProps method #25797
• [4/n][dagster-fivetran] Implement fetch_fivetran_workspace_data #25788
• [3/n][dagster-fivetran] Implement FivetranClient for rework #25756
• [2/n][dagster-fivetran] Update DagsterFivetranTranslator and related classes for rework #25751
• [1/n][dagster-fivetran] Scaffold FivetranWorkspace for rework #25750
• master

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

[graphite-app]

There appears to be a mismatch between the API call and its return value. The method makes a request for column data via _make_connector_request() but then discards that result and returns schema config data instead. To fix this, replace the two calls with a single return:

return self._make_connector_request(
    method='GET',
    endpoint=f'{connector_id}/schemas/{schema_name}/tables/{table_name}/columns'
)

This will properly return the column data that was requested.

Spotted by Graphite Reviewer

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

[graphite-app]

The string formatting appears incorrect here. The %s placeholder expects a value to substitute, but the second argument is an f-string that's already been evaluated. To properly log both the table info and exception, consider:

self._log.warning("An error occurred while fetching column metadata: %s", str(e))

or if table information is needed:

self._log.warning(
    "An error occurred while fetching column metadata for table %s: %s",
    f"{schema_source_name}.{table_source_name}",
    str(e)
)

Spotted by Graphite Reviewer

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