clarify examples

astronomer · Nov 2, 2023 · 5588e1c · 5588e1c
1 parent bbda70c
commit 5588e1c
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 24 deletions.
diff --git a/fivetran_provider_async/example_dags/example_fivetran.py b/fivetran_provider_async/example_dags/example_fivetran.py
@@ -3,7 +3,6 @@
 from airflow import DAG
 
 from fivetran_provider_async.operators import FivetranOperator
-from fivetran_provider_async.sensors import FivetranSensor
 
 default_args = {
  "owner": "Airflow",
@@ -19,18 +18,8 @@
 
 with dag:
  fivetran_sync_start = FivetranOperator(
- task_id="fivetran-task",
+ task_id="fivetran_task",
  fivetran_conn_id="fivetran_default",
  connector_id="{{ var.value.connector_id }}",
  deferrable=False,
  )
-
- fivetran_sync_wait = FivetranSensor(
- task_id="fivetran-sensor",
- fivetran_conn_id="fivetran_default",
- connector_id="{{ var.value.connector_id }}",
- poke_interval=5,
- deferrable=False,
- )
-
- fivetran_sync_start >> fivetran_sync_wait
diff --git a/fivetran_provider_async/example_dags/example_fivetran_async.py b/fivetran_provider_async/example_dags/example_fivetran_async.py
@@ -3,7 +3,6 @@
 from airflow import DAG
 
 from fivetran_provider_async.operators import FivetranOperator
-from fivetran_provider_async.sensors import FivetranSensor
 
 default_args = {
  "owner": "Airflow",
@@ -19,22 +18,20 @@
 )
 
 with dag:
+ # Both of these tasks will start a Fivetran sync,
+ # and will wait for the Fivetran sync to complete before marking
+ # the task as success.
+
+ # However, the async operator uses the triggerer instance to do this,
+ # which frees up a worker slot.
+
  fivetran_async_op = FivetranOperator(
  task_id="fivetran_async_op",
  connector_id="bronzing_largely",
  )
 
  fivetran_sync_op = FivetranOperator(
- task_id="fivetran_sync_op",
- connector_id="bronzing_largely",
- deferrable=False,
- )
-
- fivetran_async_sensor = FivetranSensor(
- task_id="fivetran_async_sensor",
- connector_id="bronzing_largely",
- poke_interval=5,
- completed_after_time="{{ task_instance.xcom_pull('fivetran_sync_op', key='return_value') }}",
+ task_id="fivetran_sync_op", connector_id="bronzing_largely", deferrable=False
  )
 
- fivetran_async_op >> fivetran_sync_op >> fivetran_async_sensor
+ fivetran_async_op >> fivetran_sync_op
diff --git a/fivetran_provider_async/example_dags/example_fivetran_xcom.py b/fivetran_provider_async/example_dags/example_fivetran_xcom.py
@@ -25,6 +25,7 @@
  task_id="fivetran-operator",
  fivetran_conn_id="fivetran_default",
  connector_id="{{ var.value.connector_id }}",
+ wait_for_completion=False,
  )
 
  delay_task = PythonOperator(task_id="delay_python_task", python_callable=lambda: time.sleep(60))