airflow_dag_fivetran_async_connector_sync_trigger.py

1import os
2from datetime import datetime, timedelta
3
4from airflow import DAG
5from fivetran_provider_async.operators import FivetranOperatorAsync
6
7default_args = {
8    "owner": "airflow",
9    "depends_on_past": False,
10    "email_on_failure": False,
11    "email_on_retry": False,
12    "retries": 1,
13    "retry_delay": timedelta(minutes=5),
14}
15
16with DAG(
17    dag_id="example_fivetran_async",
18    start_date=datetime(2021, 1, 1),
19    schedule_interval=None,
20    default_args=default_args,
21    catchup=False,
22) as dag:
23
24    # Task to trigger a Fivetran connector sync asynchronously
25    fivetran_async_task = FivetranOperatorAsync(
26        task_id="fivetran_async_task",
27        fivetran_conn_id="fivetran_default",
28        connector_id=os.getenv("FIVETRAN_CONNECTOR_ID", "your_connector_id"),
29        wait_for_completion=True,
30        poll_interval=60,
31    )
32
33    fivetran_async_task