API Changes to FivetranSensor for supporting backfills

README.md

@@ -33,32 +33,72 @@ The sensor assumes the `Conn Id` is set to `fivetran`, however if you are managi
 
 ### [Fivetran Operator Async](https://github.com/astronomer/airflow-provider-fivetran-async/tree/main/fivetran_provider_async/operators.py)
 
-`FivetranOperator` submits a Fivetran sync job and monitors it on trigger for completion.
-It requires that you specify the `connector_id` of the sync job to start. You can find `connector_id` in the Settings page of the connector you configured in the [Fivetran dashboard](https://fivetran.com/dashboard/connectors).
-
-Import into your DAG via:
 ```python
 from fivetran_provider_async.operators import FivetranOperator
 ```
 
+`FivetranOperator` submits a Fivetran sync job and monitors it on trigger for completion.
+
+`FivetranOperator` requires that you specify the `connector_id` of the Fivetran connector you wish to trigger. You can find `connector_id` in the Settings page of the connector you configured in the [Fivetran dashboard](https://fivetran.com/dashboard/connectors).
+
+The `FivetranOperator` will wait for the sync to complete so long as `wait_for_completion=True` (this is the default). It is recommended that
+you run in deferrable mode (this is also the default). If `wait_for_completion=False`, the operator will return the timestamp for the last sync.
+
+Import into your DAG via:
+
 ### [Fivetran Sensor Async](https://github.com/astronomer/airflow-provider-fivetran-async/tree/main/fivetran_provider_async/sensors.py)
 
+```python
+from fivetran_provider_async.sensors import FivetranSensor
+```
+
 `FivetranSensor` monitors a Fivetran sync job for completion.
 Monitoring with `FivetranSensor` allows you to trigger downstream processes only when the Fivetran sync jobs have completed, ensuring data consistency.
 
-
+`FivetranSensor` requires that you specify the `connector_id` of the Fivetran connector you want to wait for. You can find `connector_id` in the Settings page of the connector you configured in the [Fivetran dashboard](https://fivetran.com/dashboard/connectors).
 
 You can use multiple instances of `FivetranSensor` to monitor multiple Fivetran connectors.
 
-If used in this way,
+`FivetranSensor` is most commonly useful in two scenarios:
 
-`FivetranSensor` requires that you specify the `connector_id` of the sync job to start. You can find `connector_id` in the Settings page of the connector you configured in the [Fivetran dashboard](https://fivetran.com/dashboard/connectors).
+1. Fivetran is using a separate scheduler than the Airflow scheduler.
+2. You set `wait_for_completion=False` in the `FivetranOperator`, and you need to await the `FivetranOperator` task later. (You may want to do this if you want to arrange your DAG such that some tasks are dependent on _starting_ a sync and other tasks are dependent on _completing_ a sync).
+
+If you are doing the 1st pattern, you may find it useful to set the `completed_after_time` to `data_interval_end`, or `data_interval_end` with some buffer:
 
-Import into your DAG via:
 ```python
-from fivetran_provider_async.sensors import FivetranSensor
+fivetran_sensor = FivetranSensor(
+    task_id="wait_for_fivetran_externally_scheduled_sync",
+    connector_id="bronzing_largely",
+    poke_interval=5,
+    completed_after_time="{{ data_interval_end + macros.timedelta(minutes=1) }}",
+)
+```
+
+If you are doing the 2nd pattern, you can use XComs to pass the target completed time to the sensor:
+
+```python
+fivetran_op = FivetranOperator(
+    task_id="fivetran_sync_my_db",
+    connector_id="bronzing_largely",
+    wait_for_completion=False,
+)
+
+fivetran_sensor = FivetranSensor(
+    task_id="wait_for_fivetran_db_sync",
+    connector_id="bronzing_largely",
+    poke_interval=5,
+    completed_after_time="{{ task_instance.xcom_pull('fivetran_sync_op', key='return_value') }}",
+)
+
+fivetran_op >> fivetran_sensor
 ```
 
+You may also specify the `FivetranSensor` without a `completed_after_time`.
+In this case, the sensor will make note of when the last completed time was, and will wait for a new completed time.
+
+Import into your DAG via:
+
 ## Examples
 
 See the [**examples**](https://github.com/astronomer/airflow-provider-fivetran-async/tree/main/fivetran_provider_async/example_dags) directory for an example DAG.

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

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,
-        xcom="{{ 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

fivetran_provider_async/example_dags/example_fivetran_dbt.py

@@ -4,7 +4,6 @@
 from airflow.providers.ssh.operators.ssh import SSHOperator
 
 from fivetran_provider_async.operators import FivetranOperator
-from fivetran_provider_async.sensors import FivetranSensor
 
 default_args = {
     "owner": "Airflow",
@@ -22,29 +21,15 @@
         connector_id="{{ var.value.linkedin_connector_id }}",
     )
 
-    linkedin_sensor = FivetranSensor(
-        task_id="linkedin-sensor",
-        connector_id="{{ var.value.linkedin_connector_id }}",
-        poke_interval=600,
-    )
-
     twitter_sync = FivetranOperator(
         task_id="twitter-ads-sync",
         connector_id="{{ var.value.twitter_connector_id }}",
     )
 
-    twitter_sensor = FivetranSensor(
-        task_id="twitter-sensor",
-        connector_id="{{ var.value.twitter_connector_id }}",
-        poke_interval=600,
-    )
-
     dbt_run = SSHOperator(
         task_id="dbt_ad_reporting",
         command="cd dbt_ad_reporting ; ~/.local/bin/dbt run -m +ad_reporting",
         ssh_conn_id="dbtvm",
     )
 
-    linkedin_sync >> linkedin_sensor
-    twitter_sync >> twitter_sensor
-    [linkedin_sensor, twitter_sensor] >> dbt_run
+    [linkedin_sync, twitter_sync] >> dbt_run

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))
@@ -34,7 +35,7 @@
         fivetran_conn_id="fivetran_default",
         connector_id="{{ var.value.connector_id }}",
         poke_interval=5,
-        xcom="{{ task_instance.xcom_pull('fivetran-operator', key='return_value') }}",
+        completed_after_time="{{ task_instance.xcom_pull('fivetran-operator', key='return_value') }}",
     )
 
     fivetran_operator >> delay_task >> fivetran_sensor