Fivetran (dagster-fivetran)

This library provides a Dagster integration with Fivetran.

Assets (Fivetran API)

class dagster_fivetran.FivetranWorkspace

experimental

This API may break in future versions, even between dot releases.

This class represents a Fivetran workspace and provides utilities to interact with Fivetran APIs.

sync_and_poll

experimental

This API may break in future versions, even between dot releases.

Executes a sync and poll process to materialize Fivetran assets. This method can only be used in the context of an asset execution.

Parameters: context (AssetExecutionContext) – The execution context from within @fivetran_assets.Returns: An iterator of MaterializeResult or AssetMaterialization.

Return type: Iterator[Union[AssetMaterialization, MaterializeResult]]

class dagster_fivetran.DagsterFivetranTranslator: Translator class which converts a FivetranConnectorTableProps object into AssetSpecs. Subclass this class to implement custom logic on how to translate Fivetran content into asset spec.

@dagster_fivetran.fivetran_assets

experimental

This API may break in future versions, even between dot releases.

Create a definition for how to sync the tables of a given Fivetran connector.

Parameters:

connector_id (str) – The Fivetran Connector ID. You can retrieve this value from the
workspace (FivetranWorkspace) – The Fivetran workspace to fetch assets from.
name (Optional[str], optional) – The name of the op.
group_name (Optional[str], optional) – The name of the asset group.
dagster_fivetran_translator (Optional[DagsterFivetranTranslator], optional) – The translator to use

Examples:

Sync the tables of a Fivetran connector:

from dagster_fivetran import 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",
    name="fivetran_connector_id",
    group_name="fivetran_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},
)

Sync the tables of a Fivetran connector with a custom translator:

from dagster_fivetran import (
    DagsterFivetranTranslator,
    FivetranConnectorTableProps,
    FivetranWorkspace,
    fivetran_assets
)

import dagster as dg
from dagster._core.definitions.asset_spec import replace_attributes

class CustomDagsterFivetranTranslator(DagsterFivetranTranslator):
    def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec:
        default_spec = super().get_asset_spec(props)
        return default_spec.replace_attributes(
            key=default_spec.key.with_prefix("my_prefix"),
        )


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",
    name="fivetran_connector_id",
    group_name="fivetran_connector_id",
    workspace=fivetran_workspace,
    dagster_fivetran_translator=CustomDagsterFivetranTranslator(),
)
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},
)

dagster_fivetran.load_fivetran_asset_specs

experimental

This API may break in future versions, even between dot releases.

Returns a list of AssetSpecs representing the Fivetran content in the workspace.

Parameters:

workspace (FivetranWorkspace) – The Fivetran workspace to fetch assets from.
dagster_fivetran_translator (Optional[DagsterFivetranTranslator], optional) – The translator to use

Returns: The set of assets representing the Fivetran content in the workspace.Return type: List[AssetSpec] Examples:

Loading the asset specs for a given Fivetran workspace:

from dagster_fivetran import FivetranWorkspace, load_fivetran_asset_specs

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_specs = load_fivetran_asset_specs(fivetran_workspace)
defs = dg.Definitions(assets=[*fivetran_specs], resources=\{"fivetran": fivetran_workspace}

dagster_fivetran.build_fivetran_assets_definitions

experimental

This API may break in future versions, even between dot releases.

The list of AssetsDefinition for all connectors in the Fivetran workspace.

Parameters:

workspace (FivetranWorkspace) – The Fivetran workspace to fetch assets from.
dagster_fivetran_translator (Optional[DagsterFivetranTranslator], optional) – The translator to use

Returns: The list of AssetsDefinition for all connectors in the Fivetran workspace.Return type: List[AssetsDefinition] Examples:

Sync the tables of a Fivetran connector:

from dagster_fivetran import FivetranWorkspace, build_fivetran_assets_definitions

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 = build_fivetran_assets_definitions(workspace=workspace)

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

Sync the tables of a Fivetran connector with a custom translator:

from dagster_fivetran import (
    DagsterFivetranTranslator,
    FivetranConnectorTableProps,
    FivetranWorkspace,
     build_fivetran_assets_definitions
)

import dagster as dg
from dagster._core.definitions.asset_spec import replace_attributes

class CustomDagsterFivetranTranslator(DagsterFivetranTranslator):
    def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec:
        default_spec = super().get_asset_spec(props)
        return default_spec.replace_attributes(
            key=default_spec.key.with_prefix("my_prefix"),
        )


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 = build_fivetran_assets_definitions(
    workspace=workspace,
    dagster_fivetran_translator=CustomDagsterFivetranTranslator()
)

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

class dagster_fivetran.fivetran_event_iterator.FivetranEventIterator

A wrapper around an iterator of Fivetran events which contains additional methods for post-processing the events, such as fetching column metadata.

fetch_column_metadata

experimental

This API may break in future versions, even between dot releases.

Fetches column metadata for each table synced with the Fivetran API.

Retrieves the column schema for each destination table.

Returns: An iterator of Dagster events with column metadata attached.Return type: FivetranEventIterator

Legacy

dagster_fivetran.fivetran_resource ResourceDefinition

This resource allows users to programatically interface with the Fivetran REST API to launch syncs and monitor their progress. This currently implements only a subset of the functionality exposed by the API.

For a complete set of documentation on the Fivetran REST API, including expected response JSON schemae, see the Fivetran API Docs.

To configure this resource, we recommend using the configured method.

Examples:

from dagster import job
from dagster_fivetran import fivetran_resource

my_fivetran_resource = fivetran_resource.configured(
    \{
        "api_key": \{"env": "FIVETRAN_API_KEY"},
        "api_secret": \{"env": "FIVETRAN_API_SECRET"},
    }
)

@job(resource_defs=\{"fivetran":my_fivetran_resource})
def my_fivetran_job():
    ...

dagster_fivetran.FivetranResource ResourceDefinition: This class exposes methods on top of the Fivetran REST API.

dagster_fivetran.load_assets_from_fivetran_instance

Loads Fivetran connector assets from a configured FivetranResource instance. This fetches information about defined connectors at initialization time, and will error on workspace load if the Fivetran instance is not reachable.

Parameters:

fivetran (ResourceDefinition) – A FivetranResource configured with the appropriate connection
key_prefix (Optional[CoercibleToAssetKeyPrefix]) – A prefix for the asset keys created.
connector_to_group_fn (Optional[Callable[[str], Optional[str]]]) – Function which returns an asset
io_manager_key (Optional[str]) – The IO manager key to use for all assets. Defaults to “io_manager”.
connector_to_io_manager_key_fn (Optional[Callable[[str], Optional[str]]]) – Function which returns an
connector_filter (Optional[Callable[[FivetranConnectorMetadata], bool]]) – Optional function which takes
connector_to_asset_key_fn (Optional[Callable[[FivetranConnectorMetadata, str], AssetKey]]) – Optional function
destination_ids (Optional[List[str]]) – A list of destination IDs to fetch connectors from. If None, all destinations
poll_interval (float) – The time (in seconds) that will be waited between successive polls.
poll_timeout (Optional[float]) – The maximum time that will waited before this operation is
fetch_column_metadata (bool) – If True, will fetch column schema information for each table in the connector.

Examples:

Loading all Fivetran connectors as assets:

from dagster_fivetran import fivetran_resource, load_assets_from_fivetran_instance

fivetran_instance = fivetran_resource.configured(
    \{
        "api_key": "some_key",
        "api_secret": "some_secret",
    }
)
fivetran_assets = load_assets_from_fivetran_instance(fivetran_instance)

Filtering the set of loaded connectors:

from dagster_fivetran import fivetran_resource, load_assets_from_fivetran_instance

fivetran_instance = fivetran_resource.configured(
    \{
        "api_key": "some_key",
        "api_secret": "some_secret",
    }
)
fivetran_assets = load_assets_from_fivetran_instance(
    fivetran_instance,
    connector_filter=lambda meta: "snowflake" in meta.name,
)

dagster_fivetran.build_fivetran_assets

Build a set of assets for a given Fivetran connector.

Returns an AssetsDefinition which connects the specified asset_keys to the computation that will update them. Internally, executes a Fivetran sync for a given connector_id, and polls until that sync completes, raising an error if it is unsuccessful. Requires the use of the fivetran_resource, which allows it to communicate with the Fivetran API.

Parameters:

connector_id (str) – The Fivetran Connector ID that this op will sync. You can retrieve this
destination_tables (List[str]) – schema_name.table_name for each table that you want to be
poll_interval (float) – The time (in seconds) that will be waited between successive polls.
poll_timeout (Optional[float]) – The maximum time that will waited before this operation is
io_manager_key (Optional[str]) – The io_manager to be used to handle each of these assets.
asset_key_prefix (Optional[List[str]]) – A prefix for the asset keys inside this asset.
metadata_by_table_name (Optional[Mapping[str, RawMetadataMapping]]) – A mapping from destination
group_name (Optional[str]) – A string name used to organize multiple assets into groups. This
infer_missing_tables (bool) – If True, will create asset materializations for tables specified
op_tags (Optional[Dict[str, Any]]) – A dictionary of tags for the op that computes the asset. Frameworks may expect and
fetch_column_metadata (bool) – If True, will fetch column schema information for each table in the connector.

Examples:

Basic example:

from dagster import AssetKey, repository, with_resources

from dagster_fivetran import fivetran_resource
from dagster_fivetran.assets import build_fivetran_assets

my_fivetran_resource = fivetran_resource.configured(
    \{
        "api_key": \{"env": "FIVETRAN_API_KEY"},
        "api_secret": \{"env": "FIVETRAN_API_SECRET"},
    }
)

Attaching metadata:

fivetran_assets = build_fivetran_assets(
    connector_id="foobar",
    table_names=["schema1.table1", "schema2.table2"],
    metadata_by_table_name=\{
        "schema1.table1": \{
            "description": "This is a table that contains foo and bar",
        },
        "schema2.table2": \{
            "description": "This is a table that contains baz and quux",
        },
    },
)

dagster_fivetran.fivetran_sync_op = <dagster._core.definitions.op_definition.OpDefinition object>

Executes a Fivetran sync for a given connector_id, and polls until that sync completes, raising an error if it is unsuccessful. It outputs a FivetranOutput which contains the details of the Fivetran connector after the sync successfully completes, as well as details about which tables the sync updates.

It requires the use of the fivetran_resource, which allows it to communicate with the Fivetran API.

Examples:

from dagster import job
from dagster_fivetran import fivetran_resource, fivetran_sync_op

my_fivetran_resource = fivetran_resource.configured(
    \{
        "api_key": \{"env": "FIVETRAN_API_KEY"},
        "api_secret": \{"env": "FIVETRAN_API_SECRET"},
    }
)

sync_foobar = fivetran_sync_op.configured(\{"connector_id": "foobar"}, name="sync_foobar")

@job(resource_defs=\{"fivetran": my_fivetran_resource})
def my_simple_fivetran_job():
    sync_foobar()

@job(resource_defs=\{"fivetran": my_fivetran_resource})
def my_composed_fivetran_job():
    final_foobar_state = sync_foobar(start_after=some_op())
    other_op(final_foobar_state)

Assets (Fivetran API)​

Legacy​

Assets (Fivetran API)

Legacy