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

[maximearmstrong]

Summary & Motivation

This PR implements the fivetran_assets decorator and the build_fivetran_assets_definitions factory.

• fivetran_assets can be used to create all assets for a given Fivetran connector, i.e. one asset per table in the connector.
• build_fivetran_assets_definitions can be used to create all Fivetran assets defs, one per connector. It uses fivetran_assets.

Both the asset decorator and factory use load_fivetran_asset_specs. This is motivated by the current implementation of dagster-dbt, dagster-dlt and dagster-sling - each leverages an asset decorator that loads the asset specs by itself.

To avoid calling the Fivetran API each time load_fivetran_asset_specs is called, it is cached using functools.lru_cache. load_fivetran_asset_specs uses the state-backed defs, so reloading the code won't make additional calls to the Fivetran API, but calling load_fivetran_asset_specs N times in a script will make N calls to the Fivetran API.

The goals here are:

• make the Fivetran integration as similar as possible to the other ELT integrations by using the same patterns, eg. asset decorator
• make the user experience as simple as possible and avoid having users manage the asset specs and number of calls to the Fivetran API.

How I Tested These Changes

Additional unit tests with BK.

Changelog

[dagster-fivetran] The fivetran_assets decorator is added. It can be used with the FivetranWorkspace resource and DagsterFivetranTranslator translator to load Fivetran tables for a given connector as assets in Dagster. The build_fivetran_assets_definitions factory can be used to create assets for all the connectors in your Fivetran workspace.

[maximearmstrong]

• [11/n][dagster-fivetran] Implement materialization method in FivetranWorkspace #25961 : 3 dependent PRs (#26009 , #26025 , #26181 )
• [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 👈 (View in Graphite)
• [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.

[maximearmstrong]

Caching load_fivetran_asset_specs with functools.lru_cache requires FivetranWorkspace to be hashable.

An alternative to caching load_fivetran_asset_specs would be to call it in a FivetranWorkspace cached property.

class FivetranWorkspace(ConfigurableResource):
    ...
    @cache_property
    def asset_specs(
        self, 
        dagster_fivetran_translator: Type[DagsterFivetranTranslator] = DagsterFivetranTranslator
    ) -> Sequence[AssetSpec]:
        return load_fivetran_asset_specs(
            workspace=self, 
            dagster_fivetran_translator=dagster_fivetran_translator
        )

For the context, after this and this discussions, we decided to detach spec loading from the resources, which is why I cached load_fivetran_asset_specs instead of caching a property.

[maximearmstrong]

Tagging @benpankow @dpeng817 @schrockn @alangenfeld @OwenKephart for feedback on this specific comment and the PR description about caching the asset specs calls.

[alangenfeld]

i bias towards @cached_method on an object instance instead of @lru_cache(maxsize=None) on a top level function - I think its much easier to reason about when things are expected to be cached

[dpeng817]

we went with cached_property for the FivetranWorkspace equivalent in Airlift - I think that's likely the right move here as well.

[maximearmstrong]

Updated to a cached method in FivetranWorkspace in 501caf7

[dpeng817]

I... don't think we should add these. Reasoning:

• All other cases where we added new decorators were before we converged upon a new API design - and I think likely weren't considered holistically. The @dbt_assets decorator is an exception, but we didn't have AssetSpec yet. If we could go back and do it again, I have some doubts we would have added any of them (although there is some messy transformational stuff that it needs to do, so maybe I'm wrong)
• We can always add it later. But if we add it now we're stuck supporting it and eventually adding arguments for all the asset spec things (like group_name here). Eventually you'll need to support some sort of splitting API, etc. I just think it's better to point users towards a more genericizable solution.

I'm also not sure I understand why we need both the decorator and build_fivetran_assets_definitions? Is the idea that it just provides progressively easier starting points?

If the actual function implementations were doing more complicated stuff, maybe I would see the additional value but these are pretty simple wrappers, thus I think we should hold off.