Fivetran (dagster-fivetran)¶

Ops¶

dagster_fivetran.fivetran_sync_op = <dagster.core.definitions.op_definition.OpDefinition object>[source]¶

Config Schema:

connector_id (String)

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.

poll_interval (Float, optional)

The time (in seconds) that will be waited between successive polls.

Default Value: 10

poll_timeout (Union[Float, None], optional)

The maximum time that will waited before this operation is timed out. By default, this will never time out.

Default Value: None

yield_materializations (Bool, optional)

If True, materializations corresponding to the results of the Fivetran sync will be yielded when the op executes.

Default Value: True

asset_key_prefix (List[String], optional)

If provided and yield_materializations is True, these components will be used to prefix the generated asset keys.

Default Value: [‘fivetran’]

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)

Resources¶

dagster_fivetran.fivetran_resource ResourceDefinition[source]¶

Config Schema:

api_key (dagster.StringSource)

Fivetran API Key. You can find this value on the Fivetran settings page: https://fivetran.com/account/settings

api_secret (dagster.StringSource)

Fivetran API Secret. You can find this value on the Fivetran settings page: https://fivetran.com/account/settings

disable_schedule_on_trigger (Bool, optional)

Specifies if you would like any connector that is sync’d using this resource to be automatically taken off its Fivetran schedule.

Default Value: True

request_max_retries (Int, optional)

The maximum number of times requests to the Fivetran API should be retried before failing.

Default Value: 3

request_retry_delay (Float, optional)

Time (in seconds) to wait between each request retry.

Default Value: 0.25

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

class dagster_fivetran.FivetranResource(api_key, api_secret, disable_schedule_on_trigger=True, request_max_retries=3, request_retry_delay=0.25, log=<Logger dagster.builtin (DEBUG)>)[source]¶

This class exposes methods on top of the Fivetran REST API.

get_connector_details(connector_id)[source]¶

Gets details about a given connector from the Fivetran Connector API.

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.

Returns

Parsed json data from the response to this request

Return type

Dict[str, Any]

get_connector_sync_status(connector_id)[source]¶

Gets details about the status of the most recent Fivetran sync operation for a given 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.

Returns

Tuple representing the timestamp of the last completeded sync, if it succeeded, and the currently reported sync status.

Return type

Tuple[datetime.datetime, bool, str]

make_request(method, endpoint, data=None)[source]¶

Creates and sends a request to the desired Fivetran Connector API endpoint.

Parameters

• method (str) – The http method to use for this request (e.g. “POST”, “GET”, “PATCH”).

• endpoint (str) – The Fivetran API endpoint to send this request to.

• data (Optional[str]) – JSON-formatted data string to be included in the request.

Returns

Parsed json data from the response to this request

Return type

Dict[str, Any]

poll_sync(connector_id, initial_last_sync_completion, poll_interval=10, poll_timeout=None)[source]¶

Given a Fivetran connector and the timestamp at which the previous sync completed, poll until the next sync completes.

The previous sync completion time is necessary because the only way to tell when a sync completes is when this value changes.

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.

• initial_last_sync_completion (datetime.datetime) – The timestamp of the last completed sync (successful or otherwise) for this connector, prior to running this method.

• poll_interval (float) – The time (in seconds) that will be waited between successive polls.

• poll_timeout (float) – The maximum time that will waited before this operation is timed out. By default, this will never time out.

Returns

Parsed json data representing the API response.

Return type

Dict[str, Any]

resync_and_poll(connector_id, resync_parameters, poll_interval=10, poll_timeout=None)[source]¶

Initializes a historical resync operation for the given connector, and polls until it completes.

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.

• resync_parameters (Dict[str, List[str]]) – The payload to send to the Fivetran API. This should be a dictionary with schema names as the keys and a list of tables to resync as the values.

• poll_interval (float) – The time (in seconds) that will be waited between successive polls.

• poll_timeout (float) – The maximum time that will waited before this operation is timed out. By default, this will never time out.

Returns

Object containing details about the connector and the tables it updates

Return type

FivetranOutput

start_resync(connector_id, resync_parameters)[source]¶

Initiates a historical sync of all data for multiple schema tables within a 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.

• resync_parameters (Dict[str, List[str]]) – The resync parameters to send to the Fivetran API. An example payload can be found here: https://fivetran.com/docs/rest-api/connectors#request_6

Returns

Parsed json data representing the connector details API response after

the resync is started.

Return type

Dict[str, Any]

start_sync(connector_id)[source]¶

Initiates a sync of a 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.

Returns

Parsed json data representing the connector details API response after

the sync is started.

Return type

Dict[str, Any]

sync_and_poll(connector_id, poll_interval=10, poll_timeout=None)[source]¶

Initializes a sync operation for the given connector, and polls until it completes.

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.

• poll_interval (float) – The time (in seconds) that will be waited between successive polls.

• poll_timeout (float) – The maximum time that will waited before this operation is timed out. By default, this will never time out.

Returns

Object containing details about the connector and the tables it updates

Return type

FivetranOutput

update_connector(connector_id, properties=None)[source]¶

Updates properties of a 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.

• properties (Dict[str, Any]) – The properties to be updated. For a comprehensive list of properties, see the [Fivetran docs](https://fivetran.com/docs/rest-api/connectors#modifyaconnector).

Returns

Parsed json data representing the API response.

Return type

Dict[str, Any]

update_schedule_type(connector_id, schedule_type=None)[source]¶

Updates the schedule type property of the connector to either “auto” or “manual”.

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.

• schedule_type (Optional[str]) – Either “auto” (to turn the schedule on) or “manual” (to turn it off).

Returns

Parsed json data representing the API response.

Return type

Dict[str, Any]

Assets¶

dagster_fivetran.build_fivetran_assets(connector_id, destination_tables, poll_interval=10, poll_timeout=None, io_manager_key=None, asset_key_prefix=None)[source]¶

Build a set of assets for a given Fivetran connector.

Returns an AssetsDefintion 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]).

Examples:

from dagster import AssetKey, build_assets_job

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"},
    }
)

fivetran_assets = build_fivetran_assets(
    connector_id="foobar",
    table_names=["schema1.table1", "schema2.table2"],
])

my_fivetran_job = build_assets_job(
    "my_fivetran_job",
    assets=[fivetran_assets],
    resource_defs={"fivetran": my_fivetran_resource}
)