dagster-fivetran library

This library provides a Dagster integration with Fivetran.

Component

class dagster_fivetran.FivetranAccountComponent [source]

Loads Fivetran connectors from a given Fivetran instance as Dagster assets. Materializing these assets will trigger a sync of the Fivetran connector, enabling you to schedule Fivetran syncs using Dagster.

Example:

# defs.yaml

type: dagster_fivetran.FivetranAccountComponent
attributes:
  workspace:
    account_id: your_account_id
    api_key: "{{ env.FIVETRAN_API_KEY }}"
    api_secret: "{{ env.FIVETRAN_API_SECRET }}"
  connector_selector:
    by_name:
      - my_postgres_connector
      - my_snowflake_connector

execute [source]

Executes a Fivetran sync for the selected connector.

This method can be overridden in a subclass to customize the sync execution behavior, such as adding custom logging or handling sync results differently.

Parameters:

context – The asset execution context provided by Dagster
fivetran – The FivetranWorkspace resource used to trigger and monitor syncs

Yields: AssetMaterialization or MaterializeResult events from the Fivetran sync

Example:

Override this method to add custom logging during sync execution:

from dagster_fivetran import FivetranAccountComponent
import dagster as dg

class CustomFivetranAccountComponent(FivetranAccountComponent):
    def execute(self, context, fivetran):
        context.log.info("Starting Fivetran sync")
        yield from super().execute(context, fivetran)
        context.log.info("Fivetran sync completed successfully")

get_asset_spec [source]

Generates an AssetSpec for a given Fivetran connector table.

This method can be overridden in a subclass to customize how Fivetran connector tables are converted to Dagster asset specs. By default, it delegates to the configured DagsterFivetranTranslator.

Parameters: props – The FivetranConnectorTableProps containing information about the connector and destination table being syncedReturns: An AssetSpec that represents the Fivetran connector table as a Dagster asset

Example:

Override this method to add custom tags based on connector properties:

from dagster_fivetran import FivetranAccountComponent
import dagster as dg

class CustomFivetranAccountComponent(FivetranAccountComponent):
    def get_asset_spec(self, props):
        base_spec = super().get_asset_spec(props)
        return base_spec.replace_attributes(
            tags={
                **base_spec.tags,
                "connector_type": props.connector_type,
                "destination": props.destination_name
            }
        )

To use the Fivetran component, see the Fivetran component integration guide.

YAML configuration

When you scaffold a Fivetran component definition, the following defs.yaml configuration file will be created:

type: dagster_fivetran.FivetranAccountComponent

attributes:
  workspace:
    account_id: test_account
    api_key: '{{ env.FIVETRAN_API_KEY }}'
    api_secret: '{{ env.FIVETRAN_API_SECRET }}'

Assets (Fivetran API)

class dagster_fivetran.FivetranWorkspace [source]

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

sync_and_poll [source]

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.
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: An iterator of MaterializeResult or AssetMaterialization.

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

Examples:

Normal sync (without config):

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):

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):

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)

class dagster_fivetran.DagsterFivetranTranslator [source]: 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 [source]

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 “Setup” tab of a given connector in the Fivetran UI.
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 to convert Fivetran content into dagster.AssetSpec. Defaults to DagsterFivetranTranslator.
connector_selector_fn (Optional[ConnectorSelectorFn]) – A function that allows for filtering which Fivetran connector assets are created for.

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

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 [source]

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 to convert Fivetran content into dagster.AssetSpec. Defaults to DagsterFivetranTranslator.
connector_selector_fn (Optional[ConnectorSelectorFn]) – A function that allows for filtering which Fivetran connector assets are created for.

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 [source]

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 to convert Fivetran content into dagster.AssetSpec. Defaults to DagsterFivetranTranslator.
connector_selector_fn (Optional[ConnectorSelectorFn]) – A function that allows for filtering which Fivetran connector assets are created for.

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

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 [source]

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 [source]

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

dagster_fivetran.ConnectorSelectorFn: alias of Callable[[FivetranConnector], bool]

dagster_fivetran.build_fivetran_polling_sensor [source]

Creates a sensor that polls a Fivetran workspace for externally-triggered sync completions and emits AssetMaterialization events into Dagster’s event log.

This is useful when Fivetran connectors are run on Fivetran’s auto-schedule (not triggered by Dagster) and you want Dagster to be aware of the resulting table updates.

Parameters:

workspace (FivetranWorkspace) – The Fivetran workspace to poll.
dagster_fivetran_translator (Optional[DagsterFivetranTranslator]) – The translator to use to convert Fivetran content into AssetSpec. Defaults to DagsterFivetranTranslator.
connector_selector_fn (Optional[ConnectorSelectorFn]) – A function to filter which connectors are polled. If None, all connectors are polled.
minimum_interval_seconds (int) – The minimum interval in seconds between sensor runs. Defaults to 30.
default_sensor_status (Optional[DefaultSensorStatus]) – The default status of the sensor.
name (Optional[str]) – The name of the sensor. Defaults to fivetran_\{account_id}__sync_status_sensor.

Returns: A sensor definition.Return type: SensorDefinition

Legacy

dagster_fivetran.fivetran_resource ResourceDefinition [source]

deprecated

This API will be removed in version 0.30. Use FivetranWorkspace instead..

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 [source]: deprecated
This API will be removed in version 0.30. Use FivetranWorkspace instead..
This class exposes methods on top of the Fivetran REST API.

dagster_fivetran.load_assets_from_fivetran_instance [source]

deprecated

This API will be removed in version 0.30. Use the build_fivetran_assets_definitions factory instead..

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 details.
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 group name for a given Fivetran connector name. If None, no groups will be created. Defaults to a basic sanitization function.
io_manager_key (Optional[str]) – The IO manager key to use for all assets. Defaults to “io_manager”. Use this if all assets should be loaded from the same source, otherwise use connector_to_io_manager_key_fn.
connector_to_io_manager_key_fn (Optional[Callable[[str], Optional[str]]]) – Function which returns an IO manager key for a given Fivetran connector name. When other ops are downstream of the loaded assets, the IOManager specified determines how the inputs to those ops are loaded. Defaults to “io_manager”.
connector_filter (Optional[Callable[[FivetranConnectorMetadata], bool]]) – Optional function which takes in connector metadata and returns False if the connector should be excluded from the output assets.
connector_to_asset_key_fn (Optional[Callable[[FivetranConnectorMetadata, str], AssetKey]]) – Optional function which takes in connector metadata and a table name and returns an AssetKey for that table. Defaults to a function that generates an AssetKey matching the table name, split by “.”.
destination_ids (Optional[List[str]]) – A list of destination IDs to fetch connectors from. If None, all destinations will be polled for connectors.
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 timed out. By default, this will never time out.
fetch_column_metadata (bool) – If True, will fetch column schema information for each table in the connector. This will induce additional API calls.

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 [source]

deprecated

This API will be removed in version 0.30. Use the fivetran_assets decorator instead..

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 value from the “Setup” tab of a given connector in the Fivetran UI.
destination_tables (List[str]) – schema_name.table_name for each table that you want to be represented in the Dagster asset graph for this connection.
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 timed out. By default, this will never time out.
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. If left blank, assets will have a key of AssetKey([schema_name, table_name]).
metadata_by_table_name (Optional[Mapping[str, RawMetadataMapping]]) – A mapping from destination table name to user-supplied metadata that should be associated with the asset for that table.
group_name (Optional[str]) – A string name used to organize multiple assets into groups. This group name will be applied to all assets produced by this multi_asset.
infer_missing_tables (bool) – If True, will create asset materializations for tables specified in destination_tables even if they are not present in the Fivetran sync output. This is useful in cases where Fivetran does not sync any data for a table and therefore does not include it in the sync output API response.
op_tags (Optional[Dict[str, Any]]) – A dictionary of tags for the op that computes the asset. Frameworks may expect and require certain metadata to be attached to a op. Values that are not strings will be json encoded and must meet the criteria that json.loads(json.dumps(value)) == value.
fetch_column_metadata (bool) – If True, will fetch column schema information for each table in the connector. This will induce additional API calls.

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> [source]

deprecated

This API will be removed in version 0.30. Fivetran ops are no longer best practice and will soon be removed. Use FivetranWorkspace resource and @fivetran_asset decorator instead..

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)

Component​

YAML configuration​

Assets (Fivetran API)​

Legacy​

Component

YAML configuration

Assets (Fivetran API)

Legacy