> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Use a listener to send a Slack notification when a dataset is updated

<Info>
  This page hasn't yet been updated for Airflow 3. The concepts shown are relevant, but some code may need to be updated. If you run any examples, take care to update import statements and watch for any other breaking changes.
</Info>

[Airflow listeners](https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/listeners.html#listeners) allow you to execute custom code when certain events occur anywhere in your Airflow instance, for example when any DAG run fails or any dataset is updated.

Listeners are implemented as an [Airflow plugin](/docs/learn/using-airflow-plugins) and can contain any code. In this tutorial, you'll use a listener to send a Slack notification whenever any dataset is updated.

<Info>
  If you only need to implement notifications for specific DAGs and tasks, consider using [Airflow callbacks](/docs/learn/error-notifications-in-airflow#airflow-callbacks) instead.
</Info>

<Warning>
  The `on_dataset_created` and `on_dataset_changed` listeners are currently considered experimental and might be subject to breaking changes in future releases.
</Warning>

## Time to complete

This tutorial takes approximately 15 minutes to complete.

## Assumed knowledge

To get the most out of this tutorial, make sure you have an understanding of:

* Airflow fundamentals, such as writing DAGs and defining tasks. See [Get started with Apache Airflow](/docs/learn/get-started-with-airflow).
* Airflow plugins. See [Airflow plugins](/docs/learn/using-airflow-plugins).
* Airflow datasets. See [Datasets and data-aware scheduling in Airflow](/docs/learn/airflow-datasets).

## Prerequisites

* The [Astro CLI](/docs/cli/v1.43/get-started-cli) using [Astro Runtime](/docs/runtime/runtime-release-notes) 10+ (Airflow 2.8+).
* A Slack workspace with an [Incoming Webhook](https://api.slack.com/messaging/webhooks) configured.

## Step 1: Configure your Astro project

1. Create a new Astro project:

   ```sh wrap theme={null}
   $ mkdir astro-listener-tutorial && cd astro-listener-tutorial
   $ astro dev init
   ```

2. Add the following line to your Astro project `requirements.txt` file to install the Slack Airflow provider.

   ```text wrap theme={null}
   apache-airflow-providers-slack==8.4.0
   ```

3. Add the following environment variable to your Astro project `.env` file to create an [Airflow connection](/docs/learn/connections) to Slack. Make sure to replace `<your-slack-webhook-token>` with your own Slack webhook token in the format of `T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX`.

   ```text wrap theme={null}
   AIRFLOW_CONN_SLACK_WEBHOOK_CONN='{
       "conn_type": "slackwebhook",
       "host": "https://hooks.slack.com/services/",
       "password": "<your-slack-webhook-token>"
   }'
   ```

## Step 2: Create your listener

To define an Airflow listener, you add the code you want to execute to a relevant `@hookimpl`-decorated [listener function](https://github.com/apache/airflow/tree/main/airflow-core/src/airflow/listeners/spec). In this example, you define your code in the `on_dataset_changed` function to run whenever any dataset is updated.

1. Create a new file called `listeners_code.py` in your `plugins` folder.
2. Copy the following code into the file:

```python wrap theme={null}
from airflow.datasets import Dataset
from airflow.listeners import hookimpl
from airflow.models.taskinstance import TaskInstance
from airflow.utils.state import TaskInstanceState
from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
from sqlalchemy.orm.session import Session
from datetime import datetime

SLACK_CONN_ID = "slack_webhook_conn"

@hookimpl
def on_dataset_changed(dataset: Dataset):
    """Execute if a dataset is updated."""
    print("I am always listening for any Dataset changes and I heard that!")
    print("Posting to Slack...")
    hook = SlackWebhookHook(slack_webhook_conn_id=SLACK_CONN_ID)
    hook.send(text=f"A dataset was changed!")
    print("Done!")
    if dataset.uri == "file://include/bears":
        print("Oh! This is the bears dataset!")
        print("Bears are great :)")
        start_date = datetime.now().date()
        end_date = datetime(2024, 10, 4).date()
        days_until = (end_date - start_date).days
        print(f"Only approximately {days_until} days until fat bear week!")
```

This listener is defined using the [`on_dataset_changed` hookspec](https://github.com/apache/airflow/blob/v2-10-stable/airflow/listeners/spec/dataset.py). It posts a message to Slack whenever any dataset is updated and executes an additional print statement if the dataset that is being updated has the URI `file://include/bears`.

## Step 3: Create the listener plugin

For Airflow to recognize your listener, you need to create a [plugin](/docs/learn/using-airflow-plugins) that registers it.

1. Create a new file called `listener_plugin.py` in your `plugins` folder.

2. Copy the following code into the file:

   ```python wrap theme={null}
   from airflow.plugins_manager import AirflowPlugin
   from plugins import listeners_code

   class MyListenerPlugin(AirflowPlugin):
       name = "my_listener_plugin"
       listeners = [listeners_code]
   ```

3. If your local Airflow environment is already running, restart it to apply the changes to your plugins.

## Step 4: Create your DAG

1. In your `dags` folder, create a file called `producer_dag.py`.

2. Copy the following code into the file.

   ```python expandable wrap theme={null}
   """
   ## DAG to produce to a Dataset showcasing the on_dataset_changed listener

   This DAG will produce to a Dataset, updating it which triggers the
   on_dataset_changed listener define as an Airflow Plugin.

   The DAG also shows the difference between a Dataset and ObjectStoragePath.
   """

   from airflow.datasets import Dataset
   from airflow.decorators import dag, task
   from airflow.io.path import ObjectStoragePath
   from pendulum import datetime
   import requests


   URI = "file://include/bears"
   MY_DATASET = Dataset(URI)
   base_local = ObjectStoragePath(URI)


   @dag(
       start_date=datetime(2023, 12, 1),
       schedule="0 0 * * 0",
       catchup=False,
       doc_md=__doc__,
       tags=["on_dataset_changed listener", "2-8"],
   )
   def producer_dag():
       @task(
           outlets=[MY_DATASET],
       )
       def get_bear(base):
           r = requests.get("https://placebear.com/200/300")
           file_path = base / "bear.jpg"

           if r.status_code == 200:
               base.mkdir(parents=True, exist_ok=True)
               file_path.write_bytes(r.content)
               file_path.replace("bear.jpg")
           else:
               print(f"Failed to retrieve image. Status code: {r.status_code}")

       get_bear(base=base_local)


   producer_dag()
   ```

   This simple DAG contains one task that queries the [placebear](https://placebear.com/) API and writes the image retrieved to a local `.png` file in the `include` folder using the [Airflow object storage](/docs/learn/airflow-object-storage-tutorial) feature.
   The task produces an update to the `file://include/bears` dataset, which triggers the listener you created in [Step 2](#step-2-create-your-listener).

## Step 5: Run your DAG

1. Run `astro dev start` in your Astro project to start Airflow, then open the Airflow UI at `localhost:8080`.

2. In the Airflow UI, run the `producer_dag` DAG by clicking the play button.

3. After the DAG run completed, go to the task logs of the `get_bear` task to see print statements from your listener plugin.

   ```text wrap theme={null}
   [2023-12-17, 14:46:51 UTC] {logging_mixin.py:188} INFO - I am always listening for any dataset changes and I heard that!
   [2023-12-17, 14:46:51 UTC] {logging_mixin.py:188} INFO - Posting to Slack...
   [2023-12-17, 14:46:51 UTC] {base.py:83} INFO - Using connection ID 'slack_webhook_conn' for task execution.
   [2023-12-17, 14:46:51 UTC] {logging_mixin.py:188} INFO - Done!
   [2023-12-17, 14:46:51 UTC] {logging_mixin.py:188} INFO - Oh! This is the bears dataset!
   [2023-12-17, 14:46:51 UTC] {logging_mixin.py:188} INFO - Bears are great :)
   [2023-12-17, 14:46:51 UTC] {logging_mixin.py:188} INFO - Only approximately 292 days until fat bear week
   ```

4. Open your Slack workspace to see a new message from your webhook.

   <Frame>
     <img src="https://mintcdn.com/astronomer/VJ8or-0DggGTeulp/images/img/tutorials/airflow-listeners_slack_message.png?fit=max&auto=format&n=VJ8or-0DggGTeulp&q=85&s=0e6fe760062c694dd78bd1b2fe3e5fc6" alt="Screenshot of a Slack message sent by the webhook, saying &#x22;A dataset was changed!&#x22;" width="742" height="62" data-path="images/img/tutorials/airflow-listeners_slack_message.png" />
   </Frame>

5. (Optional) View your complimentary bear picture at `include/bears/bear.png`.

## Conclusion

Congratulations! You now know how to create an Airflow listener to run custom code whenever any dataset is updated in your whole Airflow environment. Following the same pattern you can implement listeners for [other events](https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/listeners.html#listeners), such as when any task has failed, any DAG starts running or a lifecycle event occurs.